diff --git a/hetu-common/src/main/java/io/hetu/core/common/heuristicindex/IndexCacheKey.java b/hetu-common/src/main/java/io/hetu/core/common/heuristicindex/IndexCacheKey.java index f80e49b0d..dd9214d22 100644 --- a/hetu-common/src/main/java/io/hetu/core/common/heuristicindex/IndexCacheKey.java +++ b/hetu-common/src/main/java/io/hetu/core/common/heuristicindex/IndexCacheKey.java @@ -20,19 +20,15 @@ public class IndexCacheKey { private String path; private long lastModifiedTime; - private String[] indexTypes; /** - * * @param path path to the file the index files should be read for * @param lastModifiedTime lastModifiedTime of the file, used to validate the indexes - * @param indexTypes only load specified index types, e.g. bloom */ - public IndexCacheKey(String path, long lastModifiedTime, String... indexTypes) + public IndexCacheKey(String path, long lastModifiedTime) { this.path = path; this.lastModifiedTime = lastModifiedTime; - this.indexTypes = indexTypes; } public String getPath() @@ -73,9 +69,4 @@ public class IndexCacheKey { return Objects.hash(path); } - - public String[] getIndexTypes() - { - return indexTypes; - } } diff --git a/hetu-docs/en/indexer/indexer-cli.md b/hetu-docs/en/indexer/indexer-cli.md index a8e53e687..1e5432b28 100644 --- a/hetu-docs/en/indexer/indexer-cli.md +++ b/hetu-docs/en/indexer/indexer-cli.md @@ -3,48 +3,26 @@ ## Usage -The index executable will be located under the `bin` directory in the installation. - -For example, `/bin/index` and must be executed from the `bin` directory because it uses relative paths by default. +The indexer can be utilized using the hetu-cli executable located under the `bin` directory in the installation. +To create an index you can run sql queries of the form: +```roomsql +CREATE INDEX [ IF NOT EXISTS ] index_name +USING [ BITMAP | BLOOM | MINMAX ] +ON tbl_name (col_name) +WITH ( "bloom.fpp" = '0.001', [, …] ) +WHERE predicate; ``` -Usage: index [-v] [--debug] [--disableLocking] --table= - [-c=] [--column=[,...]]... - [--partition=[,...]]... - [--type=[,...]]... - [-I=[,...]] -Using this index tool, you can CREATE, SHOW and DELETE indexes. +To show all indexes or a specific index_name: +```roomsql +SHOW INDEX; +SHOW INDEX index_name; +``` -Supported index types: BITMAP, BLOOM, MINMAX - -Supported data sources: HIVE using ORC files (must be configured in {--config}/catalog/catalog_name.properties - - - command types, e.g. create, delete, show; Note: delete command - works a column level only. - --column=[,...] - column, comma separated format for multiple columns - --debug if enabled the original data for each split will - also be written to a file alongside the index - --disableLocking by default locking is enabled at the table level; if this - is set to false, the user must ensure that the same data - is not indexed by multiple callers at the same time - (indexing different columns or partitions in parallel is - allowed) - --partition=[,...] - only create index for these partitions, comma separated - format for multiple partitions - --table=
fully qualified table name - --type=[,...] - index type, comma separated format for multiple types - (supported types: BLOOM, BITMAP, MINMAX - -c, --config= - root folder of openLooKeng etc directory (default: ../etc) - -p, --plugins=[,...] - plugins dir or file, defaults to (default: . - /hetu-heuristic-index/plugins) - -v verbose +To delete an index by name: +```roomsql +DROP INDEX index_name; ``` ## Examples @@ -52,21 +30,19 @@ Supported data sources: HIVE using ORC files (must be configured in {--config}/c ### Create index ``` shell -$ ./index -v -c ../etc --table hive.schema.table --column column1,column2 --type bloom,minmax,bitmap --partition p=part1 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", verbose=true) WHERE p=part1' ``` ### Show index ``` shell -$ ./index -v -c ../etc --table hive.schema.table show +$ ./hetu-cli --config ../etc --execute "SHOW INDEX index_name" ``` ### Delete index -*Note:* index can only be deleted at table or column level, i.e. all index types will be deleted - ``` shell -$ ./index -v -c ../etc --table hive.schema.table --column column1 delete +$ ./hetu-cli --config ../etc --execute "DELETE INDEX index_name" ``` ## Notes on resource usage @@ -76,6 +52,7 @@ $ ./index -v -c ../etc --table hive.schema.table --column column1 delete By default the default JVM MaxHeapSize will be used (`java -XX:+PrintFlagsFinal -version | grep MaxHeapSize`). For improved performance, it is recommended to increase the MaxHeapSize. This can be done by setting -Xmx value: +*Note*: This should be done before executing the hetu-cli. ``` shell export JAVA_TOOL_OPTIONS="-Xmx100G" ``` @@ -84,16 +61,16 @@ In this example the MaxHeapSize will be set to 100G. ### Indexing in parallel -If creating the index for a large table is too slow on one machine, you can create index for different partitions in parallel on different machines. This requires setting the --disableLocking flag and specifying the partition(s). For example: +If creating the index for a large table is too slow on one machine, you can create an index for different partitions in parallel on different machines. This requires setting the parallelCreation property to true and specifying the partition(s). For example: On machine 1: ``` bash -$ ./index -v ---disableLocking c ../etc --table hive.schema.table --columncolumn1,column2 --type bloom,minmax,bitmap --partition p=part1 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", parallelCreation=true) WHERE p=part1' ``` On machine 2: ``` shell -$ ./index -v ---disableLocking c ../etc --table hive.schema.table --columncolumn1,column2 --type bloom,minmax,bitmap --partition p=part2 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", parallelCreation=true) WHERE p=part2' ``` diff --git a/hetu-docs/en/indexer/overview.md b/hetu-docs/en/indexer/overview.md index 58ca8a191..4be47ca0c 100644 --- a/hetu-docs/en/indexer/overview.md +++ b/hetu-docs/en/indexer/overview.md @@ -77,7 +77,7 @@ Note that you may create multiple filesystem profiles in `etc/filesystem`, sever To write index to the indexstore specified above, just change directory to your hetu installation's `bin` folder, then run: - ./index -c --table table1 --column id --type bloom create + ./hetu-cli --config --execute 'CREATE INDEX index_name USING bloom ON table1 (column)' ### Run query diff --git a/hetu-docs/zh/indexer/indexer-cli.md b/hetu-docs/zh/indexer/indexer-cli.md index 652164aa8..0c7dfe639 100644 --- a/hetu-docs/zh/indexer/indexer-cli.md +++ b/hetu-docs/zh/indexer/indexer-cli.md @@ -3,44 +3,26 @@ ## 用法 -可执行索引文件在安装中位于`bin`目录下，例如`<安装路径>/bin/index`。因为默认使用相对路径，它必须在`bin`目录下运行。 +索引命令行接口集成于`hetu-cli`中, 在安装目录的`bin`目录下运行。 +命令的使用方式如下: +```roomsql +CREATE INDEX [ IF NOT EXISTS ] index_name +USING [ BITMAP | BLOOM | MINMAX ] +ON tbl_name (col_name) +WITH ( "bloom.fpp" = '0.001', [, …] ) +WHERE predicate; ``` -使用方法：index [-v] [--debug] [--disableLocking] --table=
- [-c=] [--column=[,...]]... - [--partition=[,...]]... - [--type=[,...]]... - [-I=[,...]] -使用此索引工具，您可以创建、显示和删除索引。 +To show all indexes or a specific index_name: +```roomsql +SHOW INDEX; +SHOW INDEX index_name; +``` -支持的索引类型如下：BITMAP, BLOOM, MINMAX - -支持的索引存储：LOCAL, HDFS (必须 在{--config}/config.properties配置 - -支持的数据源：HIVE using ORC files (必须在{--config}/catalog/catalog_name.properties配置 - - - 命令类型，如create、delete、show等；说明：delete命令只作用于列级。 - - --column=[,...] - 列，使用逗号分隔多个列 - --debug 如果启用，则每个Split的原始数据也将随索引一起写入文件 - - --disableLocking 默认锁定在表级别启用；如果设置为false，用户必须确保相同的数据没有被多个调用方同时索引（允许不同列或分区并行索引） - - --partition=[,...] - 只为这些分区创建索引，用逗号分隔多个分区 - - --table=
全量表名 - --type=[,...] - 索引类型，用逗号分隔多种类型 (支持的类型：BLOOM, BITMAP, MINMAX - -c, --config= - openLooKeng etc目录的根目录（默认为../etc） - -p, --plugins=[,...] - plugins目录或文件(默认为. /hetu-heuristic-index/plugins） - - -v verbose +To delete an index by name: +```roomsql +DROP INDEX index_name; ``` ## 示例 @@ -48,21 +30,20 @@ ### 创建索引 ``` shell -$ ./index -v -c ../etc --table hive.schema.table --column column1,column2 --type bloom,minmax,bitmap --partition p=part1 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", debugEnabled=true) WHERE p=part1' ``` ### 显示索引 ``` shell -$ ./index -v -c ../etc --table hive.schema.table show +$ ./hetu-cli --config ../etc --execute "SHOW INDEX index_name" +$ ./hetu-cli --config ../etc --execute "SHOW INDEX" ``` ### 删除索引 -*注意：* 索引只能在表或列级别删除，即所有索引类型都将被删除。 - ``` shell -$ ./index -v -c ../etc --table hive.schema.table --column column1 delete +$ ./hetu-cli --config ../etc --execute "DELETE INDEX index_name" ``` ## 资源使用说明 @@ -80,16 +61,16 @@ export JAVA_TOOL_OPTIONS="-Xmx100G" ### 并行索引 -如果在一台机器上为一个大表创建索引的速度太慢，则可以在不同的机器上并行为不同的分区创建索引。这需要设置--disableLocking标志并指定分区。例如： +如果在一台机器上为一个大表创建索引的速度太慢，则可以在不同的机器上并行为不同的分区创建索引。这需要设置parallelCreation标志并指定分区。例如： 在机器1上： ``` bash -$ ./index -v ---disableLocking c ../etc --table hive.schema.table --columncolumn1,column2 --type bloom,minmax,bitmap --partition p=part1 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", parallelCreation=true) WHERE p=part1' ``` 在机器2上： ``` shell -$ ./index -v ---disableLocking c ../etc --table hive.schema.table --columncolumn1,column2 --type bloom,minmax,bitmap --partition p=part2 create +$ ./hetu-cli --config ../etc --execute 'CREATE INDEX index_name USING bloom ON hive.schema.table (column1) WITH ("bloom.fpp"="0.01", parallelCreation=true) WHERE p=part2' ``` diff --git a/hetu-docs/zh/indexer/overview.md b/hetu-docs/zh/indexer/overview.md index 2cab7f239..a06b07362 100644 --- a/hetu-docs/zh/indexer/overview.md +++ b/hetu-docs/zh/indexer/overview.md @@ -77,7 +77,7 @@ 要创建索引，首先将工作目录cd至安装目录的`bin`文件夹，然后运行： - ./index -c --table table1 --column id --type bloom create + ./hetu-cli --config --execute 'CREATE INDEX index_name USING bloom ON table1 (column)' ### 运行语句 diff --git a/hetu-heuristic-index-cli/pom.xml b/hetu-heuristic-index-cli/pom.xml deleted file mode 100644 index 3074676eb..000000000 --- a/hetu-heuristic-index-cli/pom.xml +++ /dev/null @@ -1,175 +0,0 @@ - - - - io.hetu.core - presto-root - 316 - - 4.0.0 - - hetu-heuristic-index-cli - ${dep.hetu.version} - Indexer CLI module - - - ${project.parent.basedir} - false - false - 2.0.2 - 1.7.30 - - - - - org.slf4j - slf4j-api - ${org.slf4j.version} - - - org.slf4j - slf4j-simple - runtime - - - info.picocli - picocli - 4.0.0-alpha-3 - - - org.mockito - mockito-core - 2.28.2 - test - - - net.bytebuddy - byte-buddy-agent - - - - - io.hetu.core - hetu-heuristic-index - - - org.checkerframework - checker-qual - - - javax.activation - javax.activation-api - - - javax.el - javax.el-api - - - org.slf4j - jcl-over-slf4j - - - - - io.hetu.core - hetu-common - - - com.google.guava - guava - compile - - - org.powermock - powermock-module-testng - ${dep.powermock.version} - test - - - org.powermock - powermock-api-mockito2 - ${dep.powermock.version} - test - - - org.powermock - powermock-module-testng-common - ${dep.powermock.version} - - - org.powermock - powermock-core - ${dep.powermock.version} - - - - - org.testng - testng - - - io.hetu.core - presto-spi - - - io.hetu.core - hetu-filesystem-client - - - org.objenesis - objenesis - 3.0.1 - runtime - - - org.javassist - javassist - 3.24.0-GA - - - - - - - org.apache.maven.plugins - maven-shade-plugin - 2.3 - - - package - - shade - - - hetu-heuristic-index-cli-shaded - - - - - io.hetu.core.heuristicindex.IndexCommand - - - - - - - - maven-assembly-plugin - - - bin-tgz - package - - single - - - true - - src/main/assemblies/bin.xml - - - - - - - - \ No newline at end of file diff --git a/hetu-heuristic-index-cli/src/main/bin/index b/hetu-heuristic-index-cli/src/main/bin/index deleted file mode 100644 index 6866e4f78..000000000 --- a/hetu-heuristic-index-cli/src/main/bin/index +++ /dev/null @@ -1,14 +0,0 @@ -#!/bin/bash - -loglevel=warn - -for i in "$@" -do - if [[ $i == -v* ]] ; then - loglevel=debug - fi -done - -java -Dorg.slf4j.simpleLogger.defaultLogLevel=$loglevel\ - -jar ./hetu-heuristic-index-cli-shaded.jar\ - "$@" diff --git a/hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java b/hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java deleted file mode 100644 index f8a943417..000000000 --- a/hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java +++ /dev/null @@ -1,236 +0,0 @@ -/* - * 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. - */ - -package io.hetu.core.heuristicindex; - -import io.hetu.core.common.util.SecurePathWhiteList; -import io.prestosql.spi.heuristicindex.IndexClient; -import io.prestosql.spi.heuristicindex.IndexFactory; -import io.prestosql.spi.heuristicindex.IndexMetadata; -import io.prestosql.spi.heuristicindex.IndexWriter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import picocli.CommandLine; - -import java.io.IOException; -import java.nio.file.Paths; -import java.util.Comparator; -import java.util.List; -import java.util.Locale; -import java.util.Properties; -import java.util.concurrent.Callable; - -import static com.google.common.base.Preconditions.checkArgument; -import static io.hetu.core.heuristicindex.IndexCommandUtils.loadDataSourceProperties; -import static io.hetu.core.heuristicindex.IndexCommandUtils.loadIndexStore; -import static java.util.Objects.requireNonNull; - -/** - * Entry class for indexer - */ -@CommandLine.Command(name = "index", - description = - "\nThe Heuristic Indexer allows creating indexes on existing data and stores the index external to the original data source." + - "This means existing data can be indexed without having to rewrite the source data files. " + - "And new index types not supported by the underlying data source can be created. " + - "\n\nCurrently the Hetu Engine can utilize the indexes created by the Heuristic Indexer to perform " + - "filtering (using BLOOM or MINMAX Indexes) while scheduling splits. " + - "The engine can also perform filtering while reading ORC files (using the BITMAP Index). \n\n" + - "Using this index tool, you can CREATE, SHOW and DELETE indexes. \n\n" + - "Supported index types: BITMAP, BLOOM, MINMAX\n\n" + - "Supported index stores: LOCAL, HDFS (must be configured in {--config}/config.properties\n\n" + - "Supported data sources: HIVE (must be configured in {--config}/catalog/catalog_name.properties\n\n" + - "Notes on resource usage:\n\n" + - "By default the default JVM MaxHeapSize will be used (java -XX:+PrintFlagsFinal -version | grep MaxHeapSize).\n" + - "For improved performance, it is recommended to increase the MaxHeapSize. This can be done by setting -Xmx value:\n\n" + - "export JAVA_TOOL_OPTIONS=\"-Xmx100G\"\n\n" + - "in this example the MaxHeapSize will be set to 100G.\n\n" + - "If creating the index for a large table is too slow on one machine, you can create index for different partitions in " + - "parallel on different machines. This requires setting the --disableLocking flag.\n\n" + - "For example:\n\n" + - "One machine 1:\n" + - "$ ./index -v ---disableLocking c ../etc --table hive.schema.table --column column1,column2 --type bloom,minmax,bitmap --partition p=part1 create\n\n" + - "One machine 2:\n" + - "$ ./index -v --disableLocking -c ../etc --table hive.schema.table --column column1,column2 --type bloom,minmax,bitmap --partition p=part2 create\n\n" + - "Examples:\n\n" + - "1) Create index\n\n" + - "$ ./index -v -c ../etc --table hive.schema.table --column column1,column2 --type bloom,minmax,bitmap --partition p=part1 create\n\n" + - "2) Show index\n\n" + - "$ ./index -v -c ../etc --table hive.schema.table show\n\n" + - "3) Delete index (Note: index can only be deleted at table or column level, i.e. all index types will be deleted)\n\n" + - "$ ./index -v -c ../etc --table hive.schema.table --column column1 delete\n\n" + - "") -public class IndexCommand - implements Callable -{ - private static final Logger LOG = LoggerFactory.getLogger(IndexCommand.class); - @CommandLine.Option( - names = {"-c", "--config"}, - required = true, - description = "root folder of hetu etc directory") - String configDirPath; - @CommandLine.Option( - names = {"-t", "--table"}, - required = true, - description = "fully qualified table name") - String table; - @CommandLine.Option( - names = {"-C", "--column"}, - split = ",", - description = "column, comma separated format for multiple columns") - String[] columns; - @CommandLine.Option( - names = {"-P", "--partition"}, - split = ",", - description = "only create index for these partitions, comma separated format for multiple partitions") - String[] partitions; - @CommandLine.Option( - names = {"-T", "--type"}, - split = ",", - description = "index type, comma separated format for multiple types (supported types: BLOOM, BITMAP, MINMAX") - String[] indexTypes; - @CommandLine.Parameters( - index = "0", - description = "command types, e.g. create, delete, show;" + - " delete command works a column level") - Command command; - @CommandLine.Option( - names = {"-I", "--indexproperties"}, - split = ",", - description = "additional index properties separated by comma, for example, 'bloom.fpp=0.01,bitmap.foo=bar'") - String[] indexProps; - @CommandLine.Option( - names = {"-L", "--disableLocking"}, - description = "by default locking is enabled at the table level; if this is set to false, the user must ensure " + - "that the same data is not indexed by multiple callers at the same time (indexing different columns " + - "or partitions in parallel is allowed)") - boolean disableLocking; - @CommandLine.Option( - names = {"-d", "--debug"}, - description = "if debug is enabled the original data for each split will also be written to a file " + - "alongside the index") - boolean debugEnabled; // disabled by default - @CommandLine.Option( - names = {"-v", "--verbose"}, - description = "verbose") - boolean verbose; // disabled by default - - IndexCommand() - { - } - - public IndexCommand(String configDirPath, String table, Command command) - { - this.configDirPath = configDirPath; - this.table = table; - this.command = command; - } - - /** - * start application - * - * @param args args from commandline - */ - public static void main(String[] args) - { - CommandLine commandLine = new CommandLine(new IndexCommand()); - commandLine.setUnmatchedArgumentsAllowed(Boolean.TRUE); - commandLine.execute(args); - } - - @Override - public Void call() - throws IOException - { - // validate inputs - // security check required before using values in Path - // e.g. catalog.schema.table or dc.catalog.schema.table - checkArgument(table.matches("([\\p{Alnum}_]+\\.){2,3}[\\p{Alnum}_]+"), "Invalid table name"); - - if (columns != null) { - for (String column : columns) { - checkArgument(column.matches("[\\p{Alnum}_]+"), "Invalid column name"); - } - } - - checkArgument(!configDirPath.contains("../"), - "Config directory path must be absolute or current directory and at user workspace: " + SecurePathWhiteList.getSecurePathWhiteList().toString()); - checkArgument(Paths.get(configDirPath).toFile().exists(), "Config directory does not exist"); - - checkArgument(SecurePathWhiteList.isSecurePath(configDirPath), - "Config directory path must at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); - - IndexFactory factory = IndexCommandUtils.getIndexFactory(); - - // based on the command, different values are required - try { - Properties dsProperties = loadDataSourceProperties(table, configDirPath); - Properties ixProperties = new Properties(); - if (indexProps != null) { - for (String s : indexProps) { - if (!s.contains("=")) { - throw new IllegalArgumentException("Index properties should be like 'xx.xx=xx'"); - } - String key = s.split("=")[0]; - String val = s.split("=")[1]; - ixProperties.setProperty(key, val); - } - } - IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath); - switch (command) { - case create: - requireNonNull(indexTypes, "No index type specified for create command"); - requireNonNull(columns, "No columns specified for create command"); - IndexWriter writer = factory.getIndexWriter(dsProperties, ixProperties, indexStore.getFs(), indexStore.getRoot()); - writer.createIndex(table, columns, partitions, indexTypes, !disableLocking, debugEnabled); - break; - case delete: - IndexClient deleteClient = factory.getIndexClient(indexStore.getFs(), indexStore.getRoot()); - deleteClient.deleteIndex(table, columns); - break; - case show: - IndexClient showClient = factory.getIndexClient(indexStore.getFs(), indexStore.getRoot()); - List indexes = showClient.readSplitIndex(table); - indexes.stream() - .sorted(Comparator - .comparing(IndexMetadata::getTable) - .thenComparing(IndexMetadata::getColumn) - .thenComparing(i -> i.getIndex().getId()) - .thenComparing(IndexMetadata::getUri) - .thenComparing(IndexMetadata::getSplitStart)) - .forEach(System.out::println); - break; - default: - throw new IllegalArgumentException(String.format(Locale.ENGLISH, - "Command [%s] is not supported by the indexer", command)); - } - } - catch (IOException e) { - LOG.error("Error occurred, please check the stacktrace for details: ", e); - } - - return null; - } - - /** - * Enumeration for different Commands that indexer supports - */ - enum Command - { - create, - delete, - show - } -} diff --git a/hetu-heuristic-index/pom.xml b/hetu-heuristic-index/pom.xml index 5916d6226..927b42035 100644 --- a/hetu-heuristic-index/pom.xml +++ b/hetu-heuristic-index/pom.xml @@ -214,6 +214,43 @@ org.mockito mockito-core + 2.28.2 + test + + + org.powermock + powermock-core + 2.0.2 + test + + + org.powermock + powermock-module-testng + 2.0.2 + test + + + org.powermock + powermock-module-testng-common + 2.0.2 + test + + + org.powermock + powermock-api-mockito2 + 2.0.2 + test + + + org.objenesis + objenesis + 3.0.1 + test + + + org.javassist + javassist + 3.26.0-GA test diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java index 513c188bb..2152735c4 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java @@ -25,8 +25,6 @@ import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.heuristicindex.Index; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexMetadata; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.InputStream; @@ -47,6 +45,8 @@ import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; +import static io.hetu.core.heuristicindex.util.IndexConstants.COLUMN_DELIMITER; +import static io.hetu.core.heuristicindex.util.IndexServiceUtils.printVerboseMsg; import static java.util.Objects.requireNonNull; /** @@ -57,7 +57,6 @@ import static java.util.Objects.requireNonNull; public class HeuristicIndexClient implements IndexClient { - private static final Logger LOG = LoggerFactory.getLogger(HeuristicIndexClient.class); private static final HetuFileSystemClient LOCAL_FS_CLIENT = new HetuLocalFileSystemClient( new LocalConfig(new Properties()), Paths.get("/")); @@ -75,20 +74,22 @@ public class HeuristicIndexClient } @Override - public List readSplitIndex(String path, String... filterIndexTypes) + public List readSplitIndex(String path) throws IOException { requireNonNull(path, "no path specified"); List indexes = new LinkedList<>(); - for (Map.Entry entry : readIndexMap(path, filterIndexTypes).entrySet()) { + for (Map.Entry entry : readIndexMap(path).entrySet()) { String absolutePath = entry.getKey(); Path remainder = Paths.get(absolutePath.replaceFirst(root.toString(), "")); Path table = remainder.subpath(0, 1); remainder = Paths.get(remainder.toString().replaceFirst(table.toString(), "")); Path column = remainder.subpath(0, 1); remainder = Paths.get(remainder.toString().replaceFirst(column.toString(), "")); + Path indexType = remainder.subpath(0, 1); + remainder = Paths.get(remainder.toString().replaceFirst(indexType.toString(), "")); Path filenamePath = remainder.getFileName(); if (filenamePath == null) { @@ -97,13 +98,14 @@ public class HeuristicIndexClient remainder = remainder.getParent(); table = table.getFileName(); column = column.getFileName(); - if (remainder == null || table == null || column == null) { + indexType = indexType.getFileName(); + if (remainder == null || table == null || column == null || indexType == null) { throw new IllegalArgumentException("Split path cannot be resolved: " + path); } String filename = filenamePath.toString(); long splitStart = Long.parseLong(filename.substring(0, filename.lastIndexOf('.'))); - String timeDir = Paths.get(table.toString(), column.toString(), remainder.toString()).toString(); + String timeDir = Paths.get(table.toString(), column.toString(), indexType.toString(), remainder.toString()).toString(); long lastUpdated = getLastModified(timeDir); IndexMetadata index = new IndexMetadata( @@ -139,7 +141,7 @@ public class HeuristicIndexClient } } else { - LOG.debug("File path not valid: {}", child); + printVerboseMsg(String.format("File path not valid: %s", child)); return 0; } } @@ -149,14 +151,16 @@ public class HeuristicIndexClient } @Override - public void deleteIndex(String table, String[] columns) + public void deleteIndex(String table, String[] columns, String indexType) throws IOException { - // get the parts just to validate the table name - IndexServiceUtils.getTableParts(table); + Path toDelete = root.resolve(table).resolve(String.join(COLUMN_DELIMITER, columns)).resolve(indexType); - Path tablePath = root.resolve(table); - Lock lock = new FileBasedLock(fs, tablePath); + if (!fs.exists(toDelete)) { + return; + } + + Lock lock = new FileBasedLock(fs, toDelete.getParent()); try { Runtime.getRuntime().addShutdownHook(new Thread(() -> { lock.unlock(); @@ -169,26 +173,13 @@ public class HeuristicIndexClient })); lock.lock(); - if (columns == null) { - LOG.info("Deleted index for table {}", table); - fs.deleteRecursively(tablePath); - } - else { - for (String column : columns) { - Path indexFilePath = tablePath.resolve(column); - if (!fs.exists(indexFilePath)) { - LOG.warn("No index found for column {}", column); - } - else { - fs.deleteRecursively(indexFilePath); - LOG.info("Deleted index for column {}", column); - } - } - } + fs.deleteRecursively(toDelete); } finally { lock.unlock(); } + + return; } private boolean notDirectory(Path path) @@ -205,11 +196,10 @@ public class HeuristicIndexClient * * @param path relative path to the index file or dir, if dir, it will be searched recursively (relative to the * root uri, if one was set) - * @param filterIndexTypes only load index types matching these types, if empty or null, all types will be loaded * @return an immutable mapping from all index files read to the corresponding index that was loaded * @throws IOException */ - private Map readIndexMap(String path, String... filterIndexTypes) + private Map readIndexMap(String path) throws IOException { ImmutableMap.Builder result = ImmutableMap.builder(); @@ -224,13 +214,13 @@ public class HeuristicIndexClient try (Stream tarsOnRemote = fs.walk(absolutePath).filter(p -> p.toString().contains(".tar"))) { for (Path tarFile : (Iterable) tarsOnRemote::iterator) { Path localTmpDir = Files.createTempDirectory("tmp-index-dump-").toAbsolutePath(); - LOG.debug("Fetching index from remote filesystem to local: " + tarFile); + printVerboseMsg("Fetching index from remote filesystem to local: " + tarFile); IndexServiceUtils.unArchive(fs, LOCAL_FS_CLIENT, tarFile, localTmpDir); Path tarReadPath = Paths.get(localTmpDir.toString(), absolutePath.toString()); try (Stream children = LOCAL_FS_CLIENT.walk(tarReadPath).filter(this::notDirectory)) { for (Path child : (Iterable) children::iterator) { - LOG.debug("Processing file {}.", child); + printVerboseMsg(String.format("Processing file %s.", child)); Path childFilePath = child.getFileName(); if (childFilePath == null) { @@ -244,21 +234,6 @@ public class HeuristicIndexClient String indexType = filename.substring(filename.lastIndexOf('.') + 1).toLowerCase(Locale.ENGLISH); - if (filterIndexTypes != null && filterIndexTypes.length != 0) { - // check if indexType matches any of the indexTypes expected to be loaded - boolean found = false; - for (int i = 0; i < filterIndexTypes.length; i++) { - if (filterIndexTypes[i].equalsIgnoreCase(indexType)) { - found = true; - break; - } - } - - if (!found) { - continue; - } - } - Index index = indexTypesMap.get(indexType); if (index != null) { try { @@ -272,7 +247,8 @@ public class HeuristicIndexClient try (InputStream is = LOCAL_FS_CLIENT.newInputStream(child)) { index.load(is); } - LOG.debug("Loaded {} index from {}.", index.getId(), child); + + printVerboseMsg(String.format("Loaded %s index from %s.", index.getId(), child)); Object fileSize = LOCAL_FS_CLIENT.getAttribute(child, SupportedFileAttributes.SIZE); if (fileSize instanceof Long) { diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java index f0423c7c6..d20c066a3 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java @@ -28,8 +28,6 @@ import io.prestosql.spi.heuristicindex.DataSource; import io.prestosql.spi.heuristicindex.Index; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexFactory; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.nio.file.Path; import java.util.HashSet; @@ -37,6 +35,7 @@ import java.util.Locale; import java.util.Properties; import java.util.Set; +import static io.hetu.core.heuristicindex.util.IndexServiceUtils.printVerboseMsg; import static java.util.Objects.requireNonNull; /** @@ -48,8 +47,6 @@ import static java.util.Objects.requireNonNull; public class HeuristicIndexFactory implements IndexFactory { - private static final Logger LOG = LoggerFactory.getLogger(HeuristicIndexFactory.class); - // Add new Index and DataSources here in the future private final Set supportedIndex = ImmutableSet.of(new BloomIndex(), new MinMaxIndex(), new BitMapIndex()); private final Set supportedDataSource = ImmutableSet.of(new EmptyDataSource(), new HiveDataSource()); @@ -61,8 +58,8 @@ public class HeuristicIndexFactory @Override public HeuristicIndexWriter getIndexWriter(Properties dataSourceProps, Properties indexProps, HetuFileSystemClient fs, Path root) { - LOG.debug("dataSourceProps: {}", dataSourceProps); - LOG.debug("indexProps: {}", indexProps); + printVerboseMsg(String.format("dataSourceProps: %s", dataSourceProps)); + printVerboseMsg(String.format("indexProps: %s", indexProps)); // Load DataSource String dataSourceName = dataSourceProps.getProperty(IndexConstants.DATASTORE_TYPE_KEY); @@ -76,17 +73,16 @@ public class HeuristicIndexFactory } if (dataSource == null) { - LOG.error("DataSource not supported: {}", dataSourceName); throw new IllegalArgumentException("DataSource not supported: " + dataSourceName); } - LOG.debug("Using DataSource: {}", dataSource); + printVerboseMsg(String.format("Using DataSource: %s", dataSource)); dataSource.setProperties(dataSourceProps); // Load Index Set indices = new HashSet<>(supportedIndex); for (Index index : indices) { - LOG.debug("Using Index: {}", index); + printVerboseMsg(String.format("Using Index: %s", index)); index.setProperties(IndexServiceUtils.getPropertiesSubset( indexProps, index.getId().toLowerCase(Locale.ENGLISH) + ".")); } @@ -99,7 +95,7 @@ public class HeuristicIndexFactory { requireNonNull(root, "No root path specified"); - LOG.debug("Creating IndexClient with given filesystem client with root path {}", root); + printVerboseMsg(String.format("Creating IndexClient with given filesystem client with root path %s", root)); return new HeuristicIndexClient(new HashSet<>(supportedIndex), fs, root); } diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexWriter.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexWriter.java index 064c85220..469ff4003 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexWriter.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexWriter.java @@ -14,6 +14,8 @@ */ package io.hetu.core.heuristicindex; +import com.google.common.base.Strings; +import com.google.common.util.concurrent.AtomicDouble; import io.hetu.core.common.util.SecurePathWhiteList; import io.hetu.core.filesystem.HetuLocalFileSystemClient; import io.hetu.core.filesystem.LocalConfig; @@ -24,8 +26,6 @@ import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.heuristicindex.DataSource; import io.prestosql.spi.heuristicindex.Index; import io.prestosql.spi.heuristicindex.IndexWriter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.OutputStream; @@ -49,6 +49,7 @@ import java.util.concurrent.locks.Lock; import java.util.stream.Stream; import static com.google.common.base.Preconditions.checkArgument; +import static io.hetu.core.heuristicindex.util.IndexServiceUtils.printVerboseMsg; import static java.util.Objects.requireNonNull; /** @@ -59,7 +60,6 @@ import static java.util.Objects.requireNonNull; public class HeuristicIndexWriter implements IndexWriter { - private static final Logger LOG = LoggerFactory.getLogger(HeuristicIndexWriter.class); private static final HetuFileSystemClient LOCAL_FS_CLIENT = new HetuLocalFileSystemClient( new LocalConfig(new Properties()), Paths.get("/")); @@ -96,23 +96,27 @@ public class HeuristicIndexWriter } @Override - public void createIndex(String table, String[] columns, String[] partitions, String... indexTypes) + public void createIndex(String table, String[] columns, String[] partitions, String indexType) throws IOException { - createIndex(table, columns, partitions, indexTypes, true, false); + createIndex(table, columns, partitions, indexType, true); } @Override - public void createIndex(String table, String[] columns, String[] partitions, String[] indexTypes, boolean lockingEnabled, boolean debugEnabled) + public void createIndex(String table, String[] columns, String[] partitions, String indexType, boolean parallelCreation) throws IOException { requireNonNull(table, "no table specified"); requireNonNull(columns, "no columns specified"); - requireNonNull(indexTypes, "no index types specified"); - checkIndexTypes(indexTypes); + requireNonNull(indexType, "no index type specified"); + Index indexTypeBaseObj = indexTypesMap.get(indexType.toLowerCase(Locale.ENGLISH)); + if (indexTypeBaseObj == null) { + String msg = String.format(Locale.ENGLISH, "Index type %s not supported.", indexType); + throw new IllegalArgumentException(msg); + } - LOG.info("Creating index for: table={} columns={} partitions={}", table, Arrays.toString(columns), - partitions == null ? "all" : Arrays.toString(partitions)); + printVerboseMsg(String.format("Creating index for: table=%s columns=%s partitions=%s", table, Arrays.toString(columns), + partitions == null ? "all" : Arrays.toString(partitions))); String[] parts = IndexServiceUtils.getTableParts(table); String databaseName = parts[DATABASE_NAME_INDEX]; @@ -120,7 +124,7 @@ public class HeuristicIndexWriter Path tmpPath = Files.createTempDirectory("tmp-indexwriter-"); String strTmpPath = tmpPath.toString(); - LOG.info("Local folder to hold temp index files: " + strTmpPath); + printVerboseMsg("Local folder to hold temp index files: " + strTmpPath); Set indexedColumns = ConcurrentHashMap.newKeySet(); @@ -134,7 +138,7 @@ public class HeuristicIndexWriter "Create index temp directory path must be at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); Lock lock = null; - if (lockingEnabled) { + if (!parallelCreation) { lock = new FileBasedLock(LOCAL_FS_CLIENT, tableIndexDirPath); // cleanup hook in case regular execution is interrupted Lock finalLock = lock; @@ -150,6 +154,10 @@ public class HeuristicIndexWriter })); lock.lock(); } + AtomicDouble progress = new AtomicDouble(); + if (!IndexCommand.verbose) { + System.out.print("\rProgress: [" + Strings.repeat(" ", 48) + "] 0%"); + } // Each datasource will read the specified table's column and will use the callback when a split has been read. // The datasource will determine what a split is, for example for ORC, the datasource may read the file @@ -158,17 +166,17 @@ public class HeuristicIndexWriter // The datasource will also return the lastModified date of the split that was read try { dataSource.readSplits(databaseName, tableName, columns, partitions, - (column, values, uri, splitStart, lastModified) -> { - LOG.debug("split read: column={}; uri={}; splitOffset={}", column, uri, splitStart); + (column, values, uri, splitStart, lastModified, currProgress) -> { + printVerboseMsg(String.format("split read: column=%s; uri=%s; splitOffset=%s", column, uri, splitStart)); if (values == null || values.length == 0) { - LOG.debug("values were null or empty, skipping column={}; uri={}; splitOffset={}", column, uri, splitStart); + printVerboseMsg(String.format("values were null or empty, skipping column=%s; uri=%s; splitOffset=%s", column, uri, splitStart)); return; } // security check required before using values in a Path if (!column.matches("[\\p{Alnum}_]+")) { - LOG.warn("Invalid column name " + column); + printVerboseMsg("Invalid column name " + column); return; } try { @@ -179,13 +187,14 @@ public class HeuristicIndexWriter throw new UncheckedIOException("Get secure path list error", e); } - Path columnIndexDirPath = tableIndexDirPath.resolve(column); + // table/columns/indexType/ + Path columnAndIndexTypePath = tableIndexDirPath.resolve(column).resolve(indexType); indexedColumns.add(column); // save the indexes in a.part dir first, it will be moved later URI uriObj = URI.create(uri); String uriIndexDirPath = Paths.get( - columnIndexDirPath.toString(), uriObj.getPath()).toString(); + columnAndIndexTypePath.toString(), uriObj.getPath()).toString(); partFiles.add(uriIndexDirPath); // store the path without the part suffix uriIndexDirPath += PART_FILE_SUFFIX; // append the part suffix @@ -203,73 +212,62 @@ public class HeuristicIndexWriter } } catch (IOException e) { - LOG.error(String.format(Locale.ENGLISH, - "error writing lastModified file: %s", lastModifiedFilePath), e); throw new UncheckedIOException("error writing lastModified file: " + lastModifiedFilePath, e); } // write index files - for (String indexType : indexTypes) { - // the indexTypesMap contains all the supported index types - // the instances in the map are the "base" instances bc they have their properties set - // we need to create a new Index instance for each split and copy the properties the base has - Index indexTypeBaseObj = indexTypesMap.get(indexType.toLowerCase(Locale.ENGLISH)); - Index splitIndex; - try { - Constructor constructor = indexTypeBaseObj.getClass().getConstructor(); - splitIndex = constructor.newInstance(); - splitIndex.setProperties(indexTypeBaseObj.getProperties()); - splitIndex.setExpectedNumOfEntries(values.length); - LOG.debug("creating split index: {}", splitIndex.getId()); - } - catch (InstantiationException - | IllegalAccessException - | NoSuchMethodException - | InvocationTargetException e) { - LOG.error("unable to create instance of index: ", e); - throw new IllegalStateException("unable to create instance of index: " + indexType, e); - } + // the indexTypesMap contains all the supported index types + // the instances in the map are the "base" instances bc they have their properties set + // we need to create a new Index instance for each split and copy the properties the base has + Index splitIndex; + try { + Constructor constructor = indexTypeBaseObj.getClass().getConstructor(); + splitIndex = constructor.newInstance(); + splitIndex.setProperties(indexTypeBaseObj.getProperties()); + splitIndex.setExpectedNumOfEntries(values.length); + printVerboseMsg(String.format("creating split index: %s", splitIndex.getId())); + } + catch (InstantiationException + | IllegalAccessException + | NoSuchMethodException + | InvocationTargetException e) { + throw new IllegalStateException("unable to create instance of index: " + indexType, e); + } - String indexFileName = indexFileNamePrefix + splitIndex.getId().toLowerCase(Locale.ENGLISH); - Path indexFilePath = Paths.get(uriIndexDirPath, indexFileName); + String indexFileName = indexFileNamePrefix + splitIndex.getId().toLowerCase(Locale.ENGLISH); + Path indexFilePath = Paths.get(uriIndexDirPath, indexFileName); - splitIndex.addValues(values); + splitIndex.addValues(values); - LOG.debug("writing split index to: {}", indexFilePath); - try (OutputStream outputStream = LOCAL_FS_CLIENT.newOutputStream(indexFilePath)) { - splitIndex.persist(outputStream); - } - catch (IOException e) { - LOG.error(String.format(Locale.ENGLISH, - "error writing index file: %s", indexFilePath), e); - throw new UncheckedIOException("error writing index file: " + indexFilePath, e); - } - - if (debugEnabled) { - String dataFileName = indexFileNamePrefix + "data"; - Path dataFilePath = Paths.get(uriIndexDirPath, dataFileName); - LOG.debug("writing split data to: {}", dataFilePath); - try (OutputStream outputStream = LOCAL_FS_CLIENT.newOutputStream(dataFilePath)) { - for (int i = 0; i < values.length; i++) { - outputStream.write(values[i] == null ? "NULL".getBytes() : values[i].toString().getBytes()); - outputStream.write('\n'); + printVerboseMsg(String.format("writing split index to: %s", indexFilePath)); + try (OutputStream outputStream = LOCAL_FS_CLIENT.newOutputStream(indexFilePath)) { + splitIndex.persist(outputStream); + if (!IndexCommand.verbose) { + currProgress *= 36; + synchronized (progress) { + if (currProgress > progress.get()) { + progress.addAndGet(currProgress - progress.get()); + int bars = (int) Math.ceil(currProgress); + System.out.printf("\rProgress: [" + Strings.repeat("|", bars) + Strings.repeat(" ", 48 - bars) + "] %d%%", + bars * 100 / 48); + System.out.flush(); } } - catch (IOException e) { - LOG.error(String.format(Locale.ENGLISH, - "error writing data file: %s", dataFilePath), e); - throw new UncheckedIOException("error writing data file: " + dataFilePath, e); - } } } + catch (IOException e) { + throw new UncheckedIOException("error writing index file: " + indexFilePath, e); + } }); if (partFiles.isEmpty()) { String msg = "No index was created. Table may be empty."; - LOG.error(msg); + System.out.println(msg); throw new IllegalStateException(msg); } + double currProgress = 0; + // move all part dirs // originalDir will be something like: /tmp/indicies/catalog.schema.table/UT_test_column/UT_test // partDir will be something like: /tmp/indicies/catalog.schema.table/UT_test_column/UT_test.part @@ -294,7 +292,7 @@ public class HeuristicIndexWriter try { try (Stream tarsOnRemote = fs.walk(originalOnTarget).filter(p -> p.toString().contains(".tar"))) { for (Path tarFile : (Iterable) tarsOnRemote::iterator) { - LOG.debug("Fetching index from target filesystem to local temp: " + tarFile); + printVerboseMsg("Fetching index from target filesystem to local temp: " + tarFile); IndexServiceUtils.unArchive(fs, LOCAL_FS_CLIENT, tarFile, tmpPath); LOCAL_FS_CLIENT.createDirectories(Paths.get(strTmpPath, tarFile.getParent().toString())); Files.createFile(Paths.get(strTmpPath, tarFile.toString().replaceAll("\\.tar", ""))); @@ -317,7 +315,7 @@ public class HeuristicIndexWriter // 2. expired original if (previousLastModifiedTime != newLastModifiedTime) { if (LOCAL_FS_CLIENT.exists(originalDir)) { - LOG.debug("Removing expired index at {}.", originalDir); + printVerboseMsg(String.format("Removing expired index at %s.", originalDir)); LOCAL_FS_CLIENT.deleteRecursively(originalDir); } LOCAL_FS_CLIENT.move(partDir, originalDir); @@ -328,7 +326,7 @@ public class HeuristicIndexWriter for (Path child : (Iterable) children::iterator) { String childName = child.getFileName().toString(); Path newPath = originalDir.resolve(childName); - LOG.debug("Moving {} to {}.", child, newPath); + printVerboseMsg(String.format("Moving %s to %s.", child, newPath)); // file "lastModified=..." with same name may exist LOCAL_FS_CLIENT.deleteIfExists(newPath); LOCAL_FS_CLIENT.move(child, newPath); @@ -340,14 +338,21 @@ public class HeuristicIndexWriter } } - LOG.debug("Created index at {}.", originalDir); + printVerboseMsg(String.format("Created index at %s.", originalDir)); fs.deleteRecursively(originalOnTarget); IndexServiceUtils.archiveTar(LOCAL_FS_CLIENT, fs, originalDir, originalOnTarget); + if (!IndexCommand.verbose) { + int bars = (int) Math.ceil(progress.addAndGet(12.0 / partFiles.size())); + System.out.printf("\rProgress: [" + Strings.repeat("|", bars) + Strings.repeat(" ", 48 - bars) + "] %d%%", + bars * 100 / 48); + System.out.flush(); + currProgress++; + } } for (String indexedColumn : indexedColumns) { - LOG.info("Created index for column {}.", indexedColumn); + printVerboseMsg(String.format("Created index for column %s.", indexedColumn)); } } finally { @@ -357,23 +362,11 @@ public class HeuristicIndexWriter lock.unlock(); } - LOG.info("Deleting local tmp folder: " + strTmpPath); + printVerboseMsg("Deleting local tmp folder: " + strTmpPath); LOCAL_FS_CLIENT.deleteRecursively(tmpPath); } } - private void checkIndexTypes(String[] indexTypes) - { - for (String indexType : indexTypes) { - Index indexTypeBaseObj = indexTypesMap.get(indexType.toLowerCase(Locale.ENGLISH)); - if (indexTypeBaseObj == null) { - String msg = String.format(Locale.ENGLISH, "Index type %s not supported.", indexType); - LOG.error(msg); - throw new IllegalArgumentException(msg); - } - } - } - private void cleanPartFiles(Collection partFiles) { if (!isCleanedUp) { @@ -392,15 +385,16 @@ public class HeuristicIndexWriter } if (!failedDeletes.isEmpty()) { - LOG.warn("Failed to delete the following files, please delete them manually."); - failedDeletes.forEach(LOG::warn); + printVerboseMsg("Failed to delete the following files, please delete them manually."); + failedDeletes.forEach(IndexServiceUtils::printVerboseMsg); } isCleanedUp = true; } } - private long getLastModifiedTime(Path path) throws IOException + private long getLastModifiedTime(Path path) + throws IOException { try (Stream children = LOCAL_FS_CLIENT.list(path).filter(this::notDirectory)) { for (Path child : (Iterable) children::iterator) { @@ -413,7 +407,7 @@ public class HeuristicIndexWriter } } else { - LOG.debug("File path not valid: {}", child); + printVerboseMsg(String.format("File path not valid: %s", child)); return 0; } } diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java new file mode 100644 index 000000000..f8a8bd8ee --- /dev/null +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java @@ -0,0 +1,205 @@ +/* + * 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. + */ + +package io.hetu.core.heuristicindex; + +import io.hetu.core.common.util.SecurePathWhiteList; +import io.hetu.core.heuristicindex.util.IndexCommandUtils; +import io.prestosql.spi.heuristicindex.IndexClient; +import io.prestosql.spi.heuristicindex.IndexFactory; +import io.prestosql.spi.heuristicindex.IndexWriter; + +import java.io.IOException; +import java.nio.file.Paths; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import static com.google.common.base.Preconditions.checkArgument; +import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadDataSourceProperties; +import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadIndexStore; +import static java.util.Objects.requireNonNull; + +/** + * Entry class for indexer + */ +public class IndexCommand +{ + String configDirPath; + String table; + String[] columns; + String[] partitions; + String indexType; + String[] indexProps; + boolean parallelCreation; + public static boolean verbose; // disabled by default + String indexName; + String user; + + public IndexCommand(String configDirPath, String name, boolean verbose) + { + this.configDirPath = configDirPath; + this.indexName = name; + IndexCommand.verbose = verbose; + } + + public IndexCommand(String configDirPath, String name, boolean verbose, String user) + { + this.configDirPath = configDirPath; + this.indexName = name; + IndexCommand.verbose = verbose; + this.user = user; + } + + public IndexCommand(String configDirPath, String name, String table, String[] columns, String[] partitions, String indexType, String[] indexProps, boolean parallelCreation, + boolean verbose, String user) + { + this.configDirPath = configDirPath; + this.table = table; + this.columns = columns; + this.partitions = partitions; + this.indexType = indexType; + this.indexProps = indexProps; + this.parallelCreation = parallelCreation; + IndexCommand.verbose = verbose; + this.indexName = name; + this.user = user; + } + + public List getIndexes() + { + try { + validatePaths(); + IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath); + + if (indexName.equals("")) { + return IndexRecordManager.readAllIndexRecords(indexStore.getFs(), indexStore.getRoot()); + } + else { + List records = Collections.singletonList( + IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName)); + if (records.get(0) == null) { + return Collections.emptyList(); + } + return records; + } + } + catch (IOException e) { + e.printStackTrace(System.err); + } + return Collections.emptyList(); + } + + public void deleteIndex() + { + try { + validatePaths(); + IndexFactory factory = IndexCommandUtils.getIndexFactory(); + IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath); + IndexClient deleteClient = factory.getIndexClient(indexStore.getFs(), indexStore.getRoot()); + IndexRecordManager.IndexRecord record = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName); + if (record == null) { + System.out.printf("Index with name [%s] does not exist.%n%n", indexName); + return; + } + if (!record.user.equals("") && !record.user.equals(user)) { + System.out.printf("Index [%s] is owned by [%s]. Can't be modified with current user [%s].%n%n", indexName, record.user, user); + return; + } + deleteClient.deleteIndex(record.table, record.columns, record.indexType); + IndexRecordManager.deleteIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName); + System.out.format("Deleted index [%s].%n%n", indexName); + } + catch (IOException e) { + e.printStackTrace(System.err); + } + } + + public void createIndex() + { + try { + // validate inputs + // security check required before using values in Path + // e.g. catalog.schema.table or dc.catalog.schema.table + checkArgument(table.matches("([\\p{Alnum}_]+\\.){2,3}[\\p{Alnum}_]+"), "Invalid table name"); + if (columns != null) { + for (String column : columns) { + checkArgument(column.matches("[\\p{Alnum}_]+"), "Invalid column name"); + } + } + + validatePaths(); + IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath); + IndexFactory factory = IndexCommandUtils.getIndexFactory(); + IndexRecordManager.IndexRecord sameNameRecord = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName); + IndexRecordManager.IndexRecord sameIndexRecord = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), table, columns, indexType); + + if (sameNameRecord == null) { + if (sameIndexRecord != null) { + System.out.printf("Index with same (table,column,indexType) already exists with name [%s]%n%n", sameIndexRecord.name); + return; + } + } + else { + if (sameIndexRecord != null) { + if (!parallelCreation) { + System.out.printf("Same entry already exists. To update, please delete old index first. " + + "If this is parallel creation, add parallel creation flag to WITH%n%n"); + return; + } + } + else { + System.out.printf("Index with name [%s] already exists with different content: [%s]%n%n", indexName, sameNameRecord); + return; + } + } + + Properties dsProperties = loadDataSourceProperties(table, configDirPath); + Properties ixProperties = new Properties(); + if (indexProps != null) { + for (String s : indexProps) { + if (!s.contains("=")) { + throw new IllegalArgumentException("Index properties should be like 'xx.xx=xx'"); + } + String key = s.split("=")[0]; + String val = s.split("=")[1]; + ixProperties.setProperty(key, val); + } + } + requireNonNull(indexType, "No index type specified for create command"); + requireNonNull(columns, "No columns specified for create command"); + IndexWriter writer = factory.getIndexWriter(dsProperties, ixProperties, indexStore.getFs(), indexStore.getRoot()); + writer.createIndex(table, columns, partitions, indexType, parallelCreation); + IndexRecordManager.addIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName, user, table, columns, indexType, partitions); + if (!verbose) { + System.out.print("\n"); + } + System.out.print("\n"); + } + catch (IOException e) { + e.printStackTrace(System.err); + } + } + + private void validatePaths() + throws IOException + { + checkArgument(!configDirPath.contains("../"), + "Config directory path must be absolute or current directory and at user workspace: " + SecurePathWhiteList.getSecurePathWhiteList().toString()); + checkArgument(Paths.get(configDirPath).toFile().exists(), "Config directory does not exist"); + checkArgument(SecurePathWhiteList.isSecurePath(configDirPath), + "Config directory path must at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); + } +} diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java new file mode 100644 index 000000000..f3027856f --- /dev/null +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java @@ -0,0 +1,246 @@ +/* + * 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. + */ +package io.hetu.core.heuristicindex; + +import io.hetu.core.common.util.SecurePathWhiteList; +import io.prestosql.spi.filesystem.FileBasedLock; +import io.prestosql.spi.filesystem.HetuFileSystemClient; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; + +import static com.google.common.base.Preconditions.checkArgument; +import static io.hetu.core.heuristicindex.util.IndexConstants.COLUMN_DELIMITER; + +public class IndexRecordManager +{ + private static final String RECORD_FILE_NAME = "INDEX_RECORDS"; + + private IndexRecordManager() {} + + public static List readAllIndexRecords(HetuFileSystemClient fs, Path root) + throws IOException + { + validatePath(root); + Path recordFile = root.resolve(RECORD_FILE_NAME); + List records = new ArrayList<>(); + + if (!fs.exists(recordFile)) { + return records; + } + + try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.newInputStream(recordFile)))) { + reader.readLine(); // skip header + while (true) { + String line = reader.readLine(); + if (line == null) { + break; + } + records.add(new IndexRecord(line)); + } + } + + return records; + } + + public static IndexRecord lookUpIndexRecord(HetuFileSystemClient fs, Path root, String name) + throws IOException + { + validatePath(root); + List records = readAllIndexRecords(fs, root); + + for (IndexRecord record : records) { + if (record.name.equals(name)) { + return record; + } + } + + return null; + } + + public static IndexRecord lookUpIndexRecord(HetuFileSystemClient fs, Path root, String table, String[] columns, String indexType) + throws IOException + { + validatePath(root); + List records = readAllIndexRecords(fs, root); + + for (IndexRecord record : records) { + if (record.table.equals(table) && Arrays.equals(record.columns, columns) && record.indexType.equals(indexType)) { + return record; + } + } + + return null; + } + + /** + * Add IndexRecord into record file. If the method is called with a name that already exists, + * it will OVERWRITE the existing entry but combine the note part + */ + public static synchronized void addIndexRecord(HetuFileSystemClient fs, Path root, String name, String user, String table, String[] columns, String indexType, String... note) + throws IOException + { + validatePath(root); + // Protect root directory + FileBasedLock lock = new FileBasedLock(fs, root); + try { + lock.lock(); + List records = readAllIndexRecords(fs, root); + String noteToWrite = String.join(",", note); + Iterator iterator = records.iterator(); + while (iterator.hasNext()) { + IndexRecord record = iterator.next(); + if (name.equals(record.name)) { + noteToWrite = record.note.equals("") ? noteToWrite : record.note + "," + noteToWrite; + iterator.remove(); + } + } + records.add(new IndexRecord(name, user, table, columns, indexType, noteToWrite)); + writeIndexRecords(fs, root, records); + } + finally { + lock.unlock(); + } + } + + public static synchronized void deleteIndexRecord(HetuFileSystemClient fs, Path root, String name) + throws IOException + { + validatePath(root); + // Protect root directory + FileBasedLock lock = new FileBasedLock(fs, root); + try { + lock.lock(); + List records = readAllIndexRecords(fs, root); + records.removeIf(record -> record.name.equals(name)); + writeIndexRecords(fs, root, records); + } + finally { + lock.unlock(); + } + } + + /** + * Write the given records into the record file. This operation OVERWRITES the existing file and is NOT atomoc. + * Therefore it should only be called from lock-protected block to avoid overwriting data. + */ + private static void writeIndexRecords(HetuFileSystemClient fs, Path root, List records) + throws IOException + { + validatePath(root); + Path recordFile = root.resolve(RECORD_FILE_NAME); + + boolean writeHead = false; + try (OutputStream os = fs.newOutputStream(recordFile)) { + // Use IndexRecord to generate a special "entry" as table head so it's easier to maintain when csv format changes + String head = new IndexRecord("Name", "User", "Table", new String[] {"Columns"}, "IndexType", "Notes").toCsvRecord(); + os.write(head.getBytes()); + for (IndexRecord record : records) { + os.write(record.toCsvRecord().getBytes()); + } + } + } + + private static void validatePath(Path root) + { + try { + checkArgument(!root.toString().contains("../"), "Index store directory path must be absolute"); + checkArgument(SecurePathWhiteList.isSecurePath(root.toString()), + "Index store directory path must be at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); + } + catch (IOException e) { + throw new IllegalArgumentException("Failed to get secure path list.", e); + } + } + + public static class IndexRecord + { + public final String name; + public final String user; + public final String table; + public final String[] columns; + public final String indexType; + public final String note; + + public IndexRecord(String name, String user, String table, String[] columns, String indexType, String note) + { + this.name = name; + this.user = user == null ? "" : user; + this.table = table; + this.columns = columns; + this.indexType = indexType; + this.note = note; + } + + public IndexRecord(String csvRecord) + { + String[] records = csvRecord.split("\\t"); + this.name = records[0]; + this.user = records[1]; + this.table = records[2]; + this.columns = records[3].split(COLUMN_DELIMITER); + this.indexType = records[4]; + this.note = records.length > 5 ? records[5] : ""; + } + + public String toCsvRecord() + { + return String.format("%s\t%s\t%s\t%s\t%s\t%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType, note); + } + + @Override + public boolean equals(Object o) + { + if (this == o) { + return true; + } + if (!(o instanceof IndexRecord)) { + return false; + } + IndexRecord that = (IndexRecord) o; + return Objects.equals(name, that.name) && + Objects.equals(user, that.user) && + Objects.equals(table, that.table) && + Arrays.equals(columns, that.columns) && + Objects.equals(indexType, that.indexType); + } + + @Override + public int hashCode() + { + int result = Objects.hash(name, user, table, indexType); + result = 31 * result + Arrays.hashCode(columns); + return result; + } + + @Override + public String toString() + { + return name + "," + + user + "," + + table + "," + + "[" + String.join(",", columns) + "]," + + indexType; + } + } +} diff --git a/hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommandUtils.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java similarity index 96% rename from hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommandUtils.java rename to hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java index 254d1b0fe..27d68cadf 100644 --- a/hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommandUtils.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java @@ -12,14 +12,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.hetu.core.heuristicindex; +package io.hetu.core.heuristicindex.util; import com.google.common.collect.ImmutableSet; import io.hetu.core.common.util.SecurePathWhiteList; import io.hetu.core.filesystem.HdfsFileSystemClientFactory; import io.hetu.core.filesystem.LocalFileSystemClientFactory; -import io.hetu.core.heuristicindex.util.IndexConstants; -import io.hetu.core.heuristicindex.util.IndexServiceUtils; +import io.hetu.core.heuristicindex.HeuristicIndexFactory; import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.filesystem.HetuFileSystemClientFactory; import io.prestosql.spi.heuristicindex.IndexFactory; @@ -118,11 +117,11 @@ public class IndexCommandUtils } } - throw new IllegalArgumentException(String.format("fs.client.type '{}' has no registered factory", fsType)); + throw new IllegalArgumentException(String.format("fs.client.type '%s' has no registered factory", fsType)); } } - static class IndexStore + public static class IndexStore { final HetuFileSystemClient fs; final Path root; diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexConstants.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexConstants.java index 2ee23c0e3..1cb0f255a 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexConstants.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexConstants.java @@ -30,6 +30,8 @@ public class IndexConstants */ public static final String LAST_MODIFIED_FILE_PREFIX = "lastModified="; + public static final String COLUMN_DELIMITER = ","; + public static final String CONFIG_FILE = "config.properties"; public static final String CATALOG_CONFIGS_DIR = "catalog"; diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexServiceUtils.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexServiceUtils.java index f88b23057..177b152c6 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexServiceUtils.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexServiceUtils.java @@ -37,6 +37,7 @@ import java.util.stream.Collectors; import java.util.stream.Stream; import static com.google.common.base.Preconditions.checkArgument; +import static io.hetu.core.heuristicindex.IndexCommand.verbose; /** * Util class for creating external index. @@ -107,7 +108,8 @@ public class IndexServiceUtils * * @param file file need to be checked */ - public static void isFileExisting(File file) throws IOException + public static void isFileExisting(File file) + throws IOException { checkArgument(file.exists(), file.getCanonicalPath() + " not found"); } @@ -119,7 +121,8 @@ public class IndexServiceUtils * @return Property object which holds all properties * @throws IOException when property file does NOT exist */ - public static Properties loadProperties(File propertyFile) throws IOException + public static Properties loadProperties(File propertyFile) + throws IOException { try (InputStream is = new FileInputStream(propertyFile)) { Properties properties = new Properties(); @@ -131,7 +134,7 @@ public class IndexServiceUtils /** * get files path with a specific suffix from a path array * - * @param paths paths array + * @param paths paths array * @param suffix specific suffix * @return first path with specific suffix from that array or null if nothing found */ @@ -168,7 +171,7 @@ public class IndexServiceUtils String tableName = parts[TABLE_NAME_INDEX]; checkArgument(!tableName.isEmpty(), INVALID_TABLE_NAME_ERR_MSG); - return new String[]{catalogName, databaseName, tableName}; + return new String[] {catalogName, databaseName, tableName}; } /** @@ -252,4 +255,11 @@ public class IndexServiceUtils } } } + + public static void printVerboseMsg(String msg) + { + if (verbose) { + System.out.println(msg); + } + } } diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/datasource/hive/HdfsOrcDataSource.java b/hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/datasource/hive/HdfsOrcDataSource.java index 64f5011fe..8b7bb95e9 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/datasource/hive/HdfsOrcDataSource.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/datasource/hive/HdfsOrcDataSource.java @@ -69,6 +69,7 @@ import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.Function; import java.util.stream.Collectors; @@ -338,13 +339,15 @@ public class HdfsOrcDataSource boolean isTransactional = AcidUtils.isTransactionalTable(tableMetadata.getTable().getParameters()); boolean isFullAcid = AcidUtils.isFullAcidTable(tableMetadata.getTable().getParameters()); List files = HadoopUtil.getFiles(getFs(), tablePath, partitions, isTransactional); + AtomicLong processedFiles = new AtomicLong(); + // schedule reading of each file ExecutorService executorServices = new ThreadPoolExecutor(getConcurrency(), getConcurrency(), 0L, TimeUnit.SECONDS, new LinkedBlockingQueue(10240)); try { List jobs = files.parallelStream().map(file -> executorServices.submit(() -> { String path = file.getPath().toString(); long lastModified = file.getModificationTime(); - + double progress = (processedFiles.incrementAndGet() / ((double) files.size())); FSDataInputStream in; try { in = getFs().open(new Path(path)); @@ -359,7 +362,7 @@ public class HdfsOrcDataSource new OrcDataSourceId(path), file.getLen(), new DataSize(1, MEGABYTE), new DataSize(1, MEGABYTE), new DataSize(1, MEGABYTE), true, in, new FileFormatDataSourceStats())) { - readOrcFile(source, path, isFullAcid, columnTypes, columnNames, lastModified, callback); + readOrcFile(source, path, isFullAcid, columnTypes, columnNames, lastModified, progress, callback); } catch (Exception e) { LOG.error(String.format(ENGLISH, "Error reading file: %s. Skipping.", path), e); @@ -386,6 +389,7 @@ public class HdfsOrcDataSource Map columnTypes, Map columnNames, long lastModified, + double progress, Callback callback) throws IOException { @@ -475,7 +479,8 @@ public class HdfsOrcDataSource columnValues.toArray(new Object[0]), path, stripeInfo.getOffset(), - lastModified); + lastModified, + progress); columnValues.clear(); } } diff --git a/hetu-heuristic-index/src/main/resources/log4j.properties b/hetu-heuristic-index/src/main/resources/log4j.properties deleted file mode 100644 index c78ad41ca..000000000 --- a/hetu-heuristic-index/src/main/resources/log4j.properties +++ /dev/null @@ -1,20 +0,0 @@ -# -# 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. -# -# Root logger option -log4j.rootLogger=INFO, stdout -# Direct log messages to stdout -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.Target=System.out -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1} - %m%n diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/CsvDataSource.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/CsvDataSource.java index bd7e28a0f..e7782addf 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/CsvDataSource.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/CsvDataSource.java @@ -77,7 +77,7 @@ public class CsvDataSource List columnValues = values.get(columnNum); - callback.call(column, columnValues.toArray(new Object[0]), e.toString(), 0, e.toFile().lastModified()); + callback.call(column, columnValues.toArray(new Object[0]), e.toString(), 0, e.toFile().lastModified(), 0); } } catch (IOException ex) { diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexClient.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexClient.java index 122b531af..9029e8225 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexClient.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexClient.java @@ -19,112 +19,43 @@ import io.hetu.core.filesystem.HetuLocalFileSystemClient; import io.hetu.core.filesystem.LocalConfig; import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.heuristicindex.Index; -import org.testng.annotations.AfterTest; import org.testng.annotations.Test; import java.io.File; import java.io.IOException; import java.util.HashSet; -import java.util.LinkedList; -import java.util.List; import java.util.Properties; import java.util.Set; -import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; public class TestHeuristicIndexClient { - List testFolders = new LinkedList<>(); - @Test - public void testDeleteAllColumns() - throws IOException - { - testDeleteSelectedColumnsHelper(new String[] {"c1", "c2", "c3"}, new String[] {"c1", "c2", "c3"}); - } - - @Test - public void testDeleteSelectedColumns() - throws IOException - { - testDeleteSelectedColumnsHelper(new String[] {"c1", "c2", "c3"}, new String[] {"c1", "c3"}); - testDeleteSelectedColumnsHelper(new String[] {"c1", "c2", "c3"}, new String[] {"c1"}); - testDeleteSelectedColumnsHelper(new String[] {"c1", "c2", "c3", "abc"}, new String[] {"c3"}); - testDeleteSelectedColumnsHelper(new String[] {"c1", "c2", "c3", "abc", "def"}, new String[] {"abc", "c2"}); - } - - private void testDeleteSelectedColumnsHelper(String[] columns, String[] deleted) + public void testDeleteSelectedColumnsHelper() throws IOException { String tableName = "catalog.schema.UT_test"; - assertTrue(columns.length >= deleted.length); - String[] remained = new String[columns.length - deleted.length]; - int i = 0; - for (String c : columns) { - boolean found = false; - for (String dc : deleted) { - if (c.equals(dc)) { - found = true; - break; - } - } - if (!found) { - remained[i++] = c; - } - } - - TempFolder folder = new TempFolder(); - testFolders.add(folder); - folder.create(); - - createIndexFolderStructure(folder.getRoot(), tableName, columns); - HetuFileSystemClient fs = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), folder.getRoot().toPath()); - Set emptyIndices = new HashSet<>(); - - HeuristicIndexClient client = new HeuristicIndexClient(emptyIndices, fs, folder.getRoot().toPath()); - client.deleteIndex(tableName, deleted); - - File tableFolder = new File(folder.getRoot(), tableName); - if (remained.length == 0) { - // if all columns are deleted, no folder/files should be under the table folder - assertEquals(tableFolder.list().length, 0); - } - else { - // if there are columns left, tableFolder has to have the same number of folders of the remained columns - assertEquals(remained.length, tableFolder.list().length); - for (String remainedColumn : remained) { - // remained column should have at least one index file - File columnFolder = new File(tableFolder, remainedColumn); - assertTrue(columnFolder.list().length > 0); - } - for (String deletedColumn : deleted) { - // deleted column should not exist - File columnFolder = new File(tableName, deletedColumn); - assertFalse(columnFolder.exists()); - } - } - } - - private void createIndexFolderStructure(File folder, String table, String... columns) - throws IOException - { - File tableFolder = new File(folder, table); - assertTrue(tableFolder.mkdir()); - for (String column : columns) { - File columnFolder = new File(tableFolder, column); + try (TempFolder folder = new TempFolder()) { + // root/catalog.schema.UT_test/testColumn/bloom/testIndex.index + folder.create(); + File tableFolder = new File(folder.getRoot().getPath(), tableName); + assertTrue(tableFolder.mkdir()); + File columnFolder = new File(tableFolder, "testColumn"); assertTrue(columnFolder.mkdirs()); - assertTrue(new File(columnFolder, "testIndex.index").createNewFile()); - } - } + File indexTypeFolder = new File(columnFolder, "bloom"); + assertTrue(indexTypeFolder.mkdirs()); + assertTrue(new File(indexTypeFolder, "testIndex.index").createNewFile()); - @AfterTest - public void cleanUp() - { - for (TempFolder folder : testFolders) { - folder.close(); + HetuFileSystemClient fs = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), folder.getRoot().toPath()); + Set emptyIndices = new HashSet<>(); + + HeuristicIndexClient client = new HeuristicIndexClient(emptyIndices, fs, folder.getRoot().toPath()); + client.deleteIndex(tableName, new String[] {"testColumn"}, "bloom"); + + assertFalse(indexTypeFolder.exists()); } } } diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexFactory.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexFactory.java index bdc41952b..daba0c02c 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexFactory.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexFactory.java @@ -74,7 +74,8 @@ public class TestHeuristicIndexFactory String table = "csv.schemaName.tableName"; String[] columns = new String[] {"0", "2"}; String[] partitons = new String[] {"p=bar"}; - writer.createIndex(table, columns, partitons, "bloom", "minmax"); + writer.createIndex(table, columns, partitons, "bloom"); + writer.createIndex(table, columns, partitons, "minmax"); IndexClient client = new HeuristicIndexFactory().getIndexClient(fs, folder.getRoot().toPath()); List splits = client.readSplitIndex(table); @@ -83,9 +84,13 @@ public class TestHeuristicIndexFactory // read the index file for csv/schemaName/tableName/p=bar/000.csv, column 2 File csvFile = new File("src/test/resources/csv/schemaName/tableName/p=bar/000.csv").getCanonicalFile(); - String filePath = Paths.get("csv.schemaName.tableName", "2", csvFile.toString()).toString(); - splits = client.readSplitIndex(filePath); - // there should be 2 splits (1 csv file * 1 columns * 2 index types) + String filePathBloom = Paths.get("csv.schemaName.tableName", "2", "bloom", csvFile.toString()).toString(); + String filePathMinMax = Paths.get("csv.schemaName.tableName", "2", "minmax", csvFile.toString()).toString(); + splits = client.readSplitIndex(filePathBloom); + // Bloom only + assertEquals(1, splits.size()); + splits.addAll(client.readSplitIndex(filePathMinMax)); + // Bloom and minmax assertEquals(2, splits.size()); BloomIndex bloomIndex = null; MinMaxIndex minMaxIndex = null; diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexWriter.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexWriter.java index f12062d9d..de60e5433 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexWriter.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexWriter.java @@ -141,7 +141,7 @@ public class TestHeuristicIndexWriter public void readSplits(String schema, String table, String[] columns, String[] partitions, DataSource.Callback callback) { Object[] values = new Object[] {"test", "dsfdfs", "random"}; - callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis()); + callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis(), 0); } }; try (TempFolder folder = new TempFolder()) { @@ -164,13 +164,13 @@ public class TestHeuristicIndexWriter .filter(Files::isRegularFile).collect(Collectors.toSet()); previousFiles.forEach(f -> LOG.info(f.toString())); - writer.createIndex(tableName, new String[] {"test"}, new String[] {}, "minmax"); + writer.createIndex(tableName, new String[] {"test"}, new String[] {}, "bloom"); LOG.info("New files:"); Set newFiles = Files.walk(Paths.get(indexFolder.getAbsolutePath())) .filter(Files::isRegularFile).collect(Collectors.toSet()); newFiles.forEach(f -> LOG.info(f.toString())); - // should be the minmax index file and lastmodified file + // catalog.schema.table/UT_test_column/minmax/UT_test/lastModified=123.tar assertEquals(newFiles.size(), 1); // all files should be different @@ -206,7 +206,7 @@ public class TestHeuristicIndexWriter callback) { Object[] values = new Object[] {"test", "dsfdfs", "random"}; - callback.call("UT_test_column", values, "UT_test", 100, 123); + callback.call("UT_test_column", values, "UT_test", 100, 123, 0); } }; @@ -236,9 +236,11 @@ public class TestHeuristicIndexWriter .filter(Files::isRegularFile).collect(Collectors.toSet()); newFiles.forEach(f -> LOG.info(f.toString())); - // Expect one tar file, containing two minmax index entries - assertEquals(newFiles.size(), 1); - assertTarEntry(newFiles.iterator().next(), 2); + // two files: + // catalog.schema.table/UT_test_column/minmax/UT_test/lastModified=123.tar + // catalog.schema.table/UT_test_column/bloom/UT_test/lastModified=123.tar + assertEquals(newFiles.size(), 2); + assertTarEntry(newFiles.iterator().next(), 1); // previous files should still be there for (Path previousFile : previousFiles) { @@ -270,7 +272,7 @@ public class TestHeuristicIndexWriter callback) { Object[] values = new Object[] {"test", "dsfdfs", "random"}; - callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis()); + callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis(), 0); } }; @@ -285,7 +287,7 @@ public class TestHeuristicIndexWriter HeuristicIndexWriter writer = new HeuristicIndexWriter(ds, indices, fs, folder.getRoot().toPath()); String tableName = "catalog.schema.table"; - writer.createIndex(tableName, new String[] {"test"}, new String[] {}, new String[] {"bloom"}, true, true); + writer.createIndex(tableName, new String[] {"test"}, new String[] {}, "bloom", true); File indexFolder = new File(folder.getRoot().getAbsolutePath() + "/" + tableName); LOG.info("Previous files:"); @@ -293,9 +295,8 @@ public class TestHeuristicIndexWriter .filter(Files::isRegularFile).collect(Collectors.toSet()); files.forEach(f -> LOG.info(f.toString())); - // Expect one tar file, containing two minmax index entries assertEquals(files.size(), 1); - assertTarEntry(files.iterator().next(), 2); + assertTarEntry(files.iterator().next(), 1); } } @@ -321,7 +322,7 @@ public class TestHeuristicIndexWriter throws IOException { Object[] values = new Object[] {"test", "dsfdfs", "random"}; - callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis()); + callback.call("UT_test_column", values, "UT_test", 100, System.currentTimeMillis(), 0); } } diff --git a/hetu-heuristic-index-cli/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java similarity index 66% rename from hetu-heuristic-index-cli/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java rename to hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java index b8199c40b..22dfe2d92 100644 --- a/hetu-heuristic-index-cli/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java @@ -15,6 +15,7 @@ package io.hetu.core.heuristicindex; import io.hetu.core.common.filesystem.TempFolder; +import io.hetu.core.heuristicindex.util.IndexCommandUtils; import io.hetu.core.heuristicindex.util.IndexConstants; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexFactory; @@ -22,9 +23,7 @@ import io.prestosql.spi.heuristicindex.IndexWriter; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.testng.PowerMockTestCase; -import org.slf4j.LoggerFactory; import org.testng.annotations.Test; -import picocli.CommandLine; import java.io.File; import java.io.FileOutputStream; @@ -34,39 +33,26 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.Properties; -import static io.hetu.core.heuristicindex.IndexCommandUtils.loadDataSourceProperties; -import static io.hetu.core.heuristicindex.IndexCommandUtils.loadIndexStore; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; +import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadDataSourceProperties; +import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadIndexStore; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.powermock.api.mockito.PowerMockito.mockStatic; import static org.powermock.api.mockito.PowerMockito.when; import static org.testng.Assert.assertNotNull; -import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; -@PrepareForTest({LoggerFactory.class, IndexCommandUtils.class}) +@PrepareForTest({IndexCommandUtils.class, IndexRecordManager.class}) @PowerMockIgnore("javax.management.*") @Test(singleThreaded = true) public class TestIndexCommand extends PowerMockTestCase { - @Test - public void validateInputs() throws IOException - { - try { - IndexCommand indexCommand = new IndexCommand("/", "catalog.schema.table", IndexCommand.Command.show); - indexCommand.call(); - } - catch (IllegalArgumentException e) { - assertTrue(e.getMessage().contains("Config directory path must at user workspace")); - } - } - - @Test + @Test(expectedExceptions = RuntimeException.class) public void testCallWithEmptyConfigDirectory() throws IOException { @@ -75,14 +61,14 @@ public class TestIndexCommand File tempFile = testFolder.newFile(); assertTrue(tempFile.delete()); - String[] args = {"--config=" + tempFile.getAbsolutePath(), "--table=catalog.schema.table", - "--column=column", "--type=bloom", "create"}; + IndexCommand indexCommand = new IndexCommand(tempFile.getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null, + "bloom", null, true, false, null); - assertRuntimeException(args); + indexCommand.createIndex(); } } - @Test + @Test(expectedExceptions = RuntimeException.class) public void testCallWithNoIndexType() throws IOException { @@ -90,12 +76,13 @@ public class TestIndexCommand testFolder.create(); mockStatic(IndexCommandUtils.class); - when(IndexCommandUtils.loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); - when(IndexCommandUtils.loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); + when(loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); + when(loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); - String[] args = {"--config=" + testFolder.getRoot().getAbsolutePath(), "--table=catalog.schema.table", "--column=column", "create"}; + IndexCommand indexCommand = new IndexCommand(testFolder.getRoot().getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null, + null, null, true, false, null); - assertRuntimeException(args); + indexCommand.createIndex(); } } @@ -109,16 +96,18 @@ public class TestIndexCommand IndexFactory factory = mock(IndexFactory.class); IndexWriter writer = mock(IndexWriter.class); when(factory.getIndexWriter(any(), any(), any(), any())).thenReturn(writer); - + mockStatic(IndexRecordManager.class); + when(IndexRecordManager.readAllIndexRecords(any(), any())).thenReturn(null); mockStatic(IndexCommandUtils.class); - when(IndexCommandUtils.loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); - when(IndexCommandUtils.loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); + when(loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); + when(loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); when(IndexCommandUtils.getIndexFactory()).thenReturn(factory); - String[] args = {"--config=" + testFolder.getRoot().getAbsolutePath(), "--table=catalog.schema.table", "--column=column", "--type=bloom", "create"}; - IndexCommand.main(args); + IndexCommand indexCommand = new IndexCommand(testFolder.getRoot().getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null, + "bloom", null, false, false, null); + indexCommand.createIndex(); - verify(writer, times(1)).createIndex(any(), any(), any(), any(), eq(true), eq(false)); + verify(writer, times(1)).createIndex(any(), any(), any(), any(), eq(false)); } } @@ -132,16 +121,18 @@ public class TestIndexCommand IndexFactory factory = mock(IndexFactory.class); IndexClient client = mock(IndexClient.class); when(factory.getIndexClient(any(), any())).thenReturn(client); - + mockStatic(IndexRecordManager.class); + when(IndexRecordManager.lookUpIndexRecord(any(), any(), anyString())).thenReturn(new IndexRecordManager.IndexRecord(null, null, null, null, null, null)); mockStatic(IndexCommandUtils.class); - when(IndexCommandUtils.loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); - when(IndexCommandUtils.loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); + when(loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties()); + when(loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null)); when(IndexCommandUtils.getIndexFactory()).thenReturn(factory); - String[] args = {"--config=" + testFolder.getRoot().getAbsolutePath(), "--table=catalog.schema.table", "--column=column", "--type=bloom", "delete"}; - IndexCommand.main(args); + IndexCommand indexCommand = new IndexCommand(testFolder.getRoot().getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null, + "bloom", null, false, false, null); + indexCommand.deleteIndex(); - verify(client, times(1)).deleteIndex(any(), any()); + verify(client, times(1)).deleteIndex(any(), any(), any()); } } @@ -191,12 +182,4 @@ public class TestIndexCommand assertNotNull(factory.getIndexWriter(dsPropsRead, ixProps, is.getFs(), is.getRoot())); } } - - private void assertRuntimeException(String[] args) - { - IndexCommand myCommand = new IndexCommand(); - CommandLine commandLine = new CommandLine(myCommand); - commandLine.parseArgs(args); - assertThrows(RuntimeException.class, myCommand::call); - } } diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java new file mode 100644 index 000000000..aac609545 --- /dev/null +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java @@ -0,0 +1,116 @@ +/* + * 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. + */ +package io.hetu.core.heuristicindex; + +import io.hetu.core.common.filesystem.TempFolder; +import io.hetu.core.filesystem.HetuLocalFileSystemClient; +import io.hetu.core.filesystem.LocalConfig; +import io.prestosql.spi.filesystem.HetuFileSystemClient; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.file.Paths; +import java.util.Properties; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; + +public class TestIndexRecordManager +{ + private static final HetuFileSystemClient FILE_SYSTEM_CLIENT = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get("/")); + + @Test + public void testDelete() + throws IOException + { + try (TempFolder folder = new TempFolder()) { + folder.create(); + IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1"); + IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1"); + assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1")); + assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 2); + + // Delete 1 + IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1"); + assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1")); + assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2")); + assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 1); + + // Delete 1 again + IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1"); + assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1")); + assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2")); + assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 1); + + // Delete 2 + IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2"); + assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2")); + assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 0); + } + } + + @Test + public void testAddAndLookUp() + throws IOException, IllegalAccessException + { + testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ""); + testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn", "testColumn2"}, "minmax", ""); + testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "12"); + testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "12", "123"); + } + + @Test(expectedExceptions = AssertionError.class) + public void testAddAndLookUpDifferentNotes() + throws IOException, IllegalAccessException + { + try (TempFolder folder = new TempFolder()) { + folder.create(); + IndexRecordManager.IndexRecord expected = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ""); + IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1"); + IndexRecordManager.IndexRecord actual = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "testName"); + assertIndexRecordFullyEqual(actual, expected); + } + } + + private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, String... note) + throws IOException, IllegalAccessException + { + try (TempFolder folder = new TempFolder()) { + folder.create(); + IndexRecordManager.IndexRecord expected = new IndexRecordManager.IndexRecord(name, user, table, columns, indexType, String.join(",", note)); + IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), name, user, table, columns, indexType, note); + + IndexRecordManager.IndexRecord actual1 = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), name); + assertNotNull(actual1); + assertIndexRecordFullyEqual(actual1, expected); + + IndexRecordManager.IndexRecord actual2 = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), table, columns, indexType); + assertNotNull(actual2); + assertIndexRecordFullyEqual(actual2, expected); + } + } + + // Compare two IndexRecord objects and assert all fields are equal. + // Unlike the equals() method of IndexRecord, this method compares ALL fields for testing. + private void assertIndexRecordFullyEqual(IndexRecordManager.IndexRecord actual, IndexRecordManager.IndexRecord expected) + throws IllegalAccessException + { + for (Field field : actual.getClass().getDeclaredFields()) { + assertEquals(field.get(actual), field.get(expected)); + } + } +} diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/datasource/hive/TestHiveDataSource.java b/hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/datasource/hive/TestHiveDataSource.java index a2a7ddc02..60295df8a 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/datasource/hive/TestHiveDataSource.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/datasource/hive/TestHiveDataSource.java @@ -166,7 +166,7 @@ public class TestHiveDataSource Map readColumnClasses = new ConcurrentHashMap<>(); testSource.readSplits(DockerizedHive.DATABASE_NAME, DockerizedHive.TYPES_TABLE_NAME, expectedColumnClasses.keySet().toArray(new String[0]), null, - (column, values, uri, splitStart, lastModified) -> readColumnClasses.putIfAbsent(column, values[0].getClass())); + (column, values, uri, splitStart, lastModified, progress) -> readColumnClasses.putIfAbsent(column, values[0].getClass())); assertEquals(readColumnClasses, expectedColumnClasses); } @@ -186,7 +186,7 @@ public class TestHiveDataSource Map readColumnClasses = new ConcurrentHashMap<>(); testSource.readSplits(DockerizedHive.DATABASE_NAME, DockerizedHive.TYPES_TABLE_NAME, new String[] {unsupportedColumn}, null, - (column, values, uri, splitStart, lastModified) -> readColumnClasses.putIfAbsent(column, values[0].getClass())); + (column, values, uri, splitStart, lastModified, progress) -> readColumnClasses.putIfAbsent(column, values[0].getClass())); } } @@ -209,7 +209,7 @@ public class TestHiveDataSource FileSystem fs = FileSystem.get(conf); long totalRows = getTotalRows(fs, tableUrl); - source.readSplits(databaseName, tableName, columns, partitions, (column, values, uri, splitStart, lastModified) -> { + source.readSplits(databaseName, tableName, columns, partitions, (column, values, uri, splitStart, lastModified, progress) -> { rowCounter.add(values.length); splitCounter.increment(); }); diff --git a/hetu-server/src/main/provisio/hetu.xml b/hetu-server/src/main/provisio/hetu.xml index 004a7d6a7..775d16573 100644 --- a/hetu-server/src/main/provisio/hetu.xml +++ b/hetu-server/src/main/provisio/hetu.xml @@ -12,8 +12,8 @@ - - hetu-heuristic-index-cli-shaded.jar + + hetu-cli-010-executable.jar @@ -22,7 +22,7 @@ - + diff --git a/pom.xml b/pom.xml index 977f8e26f..8d498164b 100644 --- a/pom.xml +++ b/pom.xml @@ -126,7 +126,6 @@ hetu-oracle hetu-vdm hetu-heuristic-index - hetu-heuristic-index-cli hetu-datacenter hetu-hana hetu-listener diff --git a/presto-cli/pom.xml b/presto-cli/pom.xml index 28fb0147d..31563b95a 100644 --- a/presto-cli/pom.xml +++ b/presto-cli/pom.xml @@ -33,7 +33,7 @@ org.antlr antlr4-runtime - + io.airlift airline @@ -118,6 +118,29 @@ jackson-core + + io.hetu.core + hetu-heuristic-index + + + javax.el + javax.el-api + + + org.glassfish + javax.el + + + org.slf4j + slf4j-api + + + org.checkerframework + checker-qual + + + + org.testng @@ -148,6 +171,7 @@ true executable + ${main-class} @@ -158,20 +182,21 @@ - - org.skife.maven - really-executable-jar-maven-plugin - - -Xmx1G - executable - + maven-assembly-plugin + bin-tgz package - really-executable-jar + single + + true + + src/main/assemblies/bin.xml + + diff --git a/hetu-heuristic-index-cli/src/main/assemblies/bin.xml b/presto-cli/src/main/assemblies/bin.xml similarity index 100% rename from hetu-heuristic-index-cli/src/main/assemblies/bin.xml rename to presto-cli/src/main/assemblies/bin.xml diff --git a/presto-cli/src/main/bin/hetu-cli b/presto-cli/src/main/bin/hetu-cli new file mode 100755 index 000000000..98e84841d --- /dev/null +++ b/presto-cli/src/main/bin/hetu-cli @@ -0,0 +1,4 @@ +#!/bin/bash + +java -jar "$(dirname "$0")/hetu-cli-010-executable.jar"\ + "$@" \ No newline at end of file diff --git a/presto-cli/src/main/java/io/prestosql/cli/ClientOptions.java b/presto-cli/src/main/java/io/prestosql/cli/ClientOptions.java index 8f8219684..f7d86c5e4 100644 --- a/presto-cli/src/main/java/io/prestosql/cli/ClientOptions.java +++ b/presto-cli/src/main/java/io/prestosql/cli/ClientOptions.java @@ -149,6 +149,9 @@ public class ClientOptions @Option(name = "--ignore-errors", title = "ignore errors", description = "Continue processing in batch mode when an error occurs (default is to exit immediately)") public boolean ignoreErrors; + @Option(name = {"-c", "--config"}, title = "root folder of hetu etc directory (default: \"../etc\")", description = "root folder of hetu etc directory (default: \"../etc\")") + public String configDirPath = "../etc"; + public enum OutputFormat { ALIGNED, diff --git a/presto-cli/src/main/java/io/prestosql/cli/Console.java b/presto-cli/src/main/java/io/prestosql/cli/Console.java index e8e96f671..c50daf051 100644 --- a/presto-cli/src/main/java/io/prestosql/cli/Console.java +++ b/presto-cli/src/main/java/io/prestosql/cli/Console.java @@ -14,15 +14,31 @@ package io.prestosql.cli; import com.google.common.base.CharMatcher; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import io.airlift.airline.Command; import io.airlift.airline.HelpOption; import io.airlift.log.Logging; import io.airlift.log.LoggingConfiguration; import io.airlift.units.Duration; +import io.hetu.core.heuristicindex.IndexCommand; +import io.hetu.core.heuristicindex.IndexRecordManager; import io.prestosql.client.ClientSelectedRole; import io.prestosql.client.ClientSession; +import io.prestosql.client.ClientTypeSignature; +import io.prestosql.client.Column; +import io.prestosql.client.ErrorLocation; +import io.prestosql.sql.parser.ParsingException; +import io.prestosql.sql.parser.SqlParser; import io.prestosql.sql.parser.StatementSplitter; +import io.prestosql.sql.tree.ComparisonExpression; +import io.prestosql.sql.tree.CreateIndex; +import io.prestosql.sql.tree.DropIndex; +import io.prestosql.sql.tree.Expression; +import io.prestosql.sql.tree.Identifier; +import io.prestosql.sql.tree.LogicalBinaryExpression; +import io.prestosql.sql.tree.Property; +import io.prestosql.sql.tree.ShowIndex; import org.jline.reader.EndOfFileException; import org.jline.reader.History; import org.jline.reader.LineReaderBuilder; @@ -35,15 +51,23 @@ import javax.inject.Inject; import java.io.File; import java.io.IOException; import java.io.PrintStream; +import java.io.StringWriter; import java.io.UncheckedIOException; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import java.util.stream.Stream; import static com.google.common.base.CharMatcher.whitespace; import static com.google.common.base.Preconditions.checkState; @@ -56,6 +80,7 @@ import static com.google.common.util.concurrent.Uninterruptibles.awaitUninterrup import static io.prestosql.cli.Help.getHelpText; import static io.prestosql.cli.QueryPreprocessor.preprocessQuery; import static io.prestosql.client.ClientSession.stripTransactionId; +import static io.prestosql.client.ClientStandardTypes.VARCHAR; import static io.prestosql.sql.parser.StatementSplitter.Statement; import static io.prestosql.sql.parser.StatementSplitter.isEmptyStatement; import static java.lang.String.format; @@ -74,6 +99,7 @@ public class Console private static final String PROMPT_NAME = "lk"; private static final Duration EXIT_DELAY = new Duration(3, SECONDS); + private static final Pattern createIndexPattern = Pattern.compile("^(drop|create|show)\\s*index.*", Pattern.CASE_INSENSITIVE); @Inject public HelpOption helpOption; @@ -89,7 +115,6 @@ public class Console ClientSession session = clientOptions.toClientSession(); boolean hasQuery = !isNullOrEmpty(clientOptions.execute); boolean isFromFile = !isNullOrEmpty(clientOptions.file); - initializeLogging(clientOptions.logLevelsFile); String query = clientOptions.execute; @@ -140,6 +165,11 @@ public class Console Optional.ofNullable(clientOptions.krb5CredentialCachePath), !clientOptions.krb5DisableRemoteServiceHostnameCanonicalization)) { if (hasQuery) { + if (createIndexPattern.matcher(query).matches()) { + executeHeuristicIndexQuery(query.substring(0, query.length() - 1), exiting, queryRunner); + return true; + } + return executeCommand( queryRunner, exiting, @@ -158,6 +188,21 @@ public class Console } } + private void executeHeuristicIndexQuery(String query, AtomicBoolean exiting, QueryRunner queryRunner) + { + switch (query.split(" ", 2)[0].toLowerCase(ENGLISH)) { + case "create": + createIndexCommand(query, queryRunner, exiting); + break; + case "show": + showIndexCommand(query, queryRunner, exiting); + break; + case "drop": + deleteIndexCommand(query, queryRunner, exiting); + break; + } + } + private String getPassword() { checkState(clientOptions.user != null, "Username must be specified along with password"); @@ -186,7 +231,158 @@ public class Console } } - private static void runConsole(QueryRunner queryRunner, AtomicBoolean exiting) + private void deleteIndexCommand(String query, QueryRunner queryRunner, AtomicBoolean exiting) + { + boolean hasAuthenticated = executeCommand( + queryRunner, + exiting, + "show catalogs;", + ClientOptions.OutputFormat.NULL, + false, + false); + + if (hasAuthenticated) { + SqlParser parser = new SqlParser(); + try { + DropIndex deleteIndex = (DropIndex) parser.createStatement(query); + IndexCommand command = new IndexCommand(clientOptions.configDirPath, deleteIndex.getIndexName().toString(), + false, clientOptions.user); + command.deleteIndex(); + } + catch (ParsingException e) { + System.out.println(e.getMessage()); + Query.renderErrorLocation(query, new ErrorLocation(e.getLineNumber(), e.getColumnNumber()), System.out); + } + catch (IllegalArgumentException e) { + System.out.println(e.getMessage()); + } + } + } + + private void showIndexCommand(String query, QueryRunner queryRunner, AtomicBoolean exiting) + { + boolean hasAuthenticated = executeCommand( + queryRunner, + exiting, + "show catalogs;", + ClientOptions.OutputFormat.NULL, + false, + false); + + if (hasAuthenticated) { + SqlParser parser = new SqlParser(); + try { + ShowIndex showIndex = (ShowIndex) parser.createStatement(query); + IndexCommand command = new IndexCommand(clientOptions.configDirPath, (showIndex.getIndexName() == null ? "" : showIndex.getIndexName().toString()), + false); + + List columns = ImmutableList.builder() + .add(new Column("Index Name", VARCHAR, new ClientTypeSignature(VARCHAR))) + .add(new Column("User", VARCHAR, new ClientTypeSignature(VARCHAR))) + .add(new Column("Table Name", VARCHAR, new ClientTypeSignature(VARCHAR))) + .add(new Column("Column Name", VARCHAR, new ClientTypeSignature(VARCHAR))) + .add(new Column("Index Type", VARCHAR, new ClientTypeSignature(VARCHAR))) + .add(new Column("Partitions", VARCHAR, new ClientTypeSignature(VARCHAR))) + .build(); + List records = command.getIndexes(); + + List> rows = new ArrayList<>(); + for (IndexRecordManager.IndexRecord v : records) { + List strings = Arrays.asList(v.name, v.user, v.table, String.join(",", v.columns), v.indexType, v.note.replaceAll("(.{70})", "$0\n")); + rows.add(strings); + } + StringWriter writer = new StringWriter(); + OutputPrinter printer = new AlignedTablePrinter(columns, writer); + printer.printRows(rows, true); + printer.finish(); + + System.out.println(writer.getBuffer().toString()); + } + catch (ParsingException e) { + System.out.println(e.getMessage()); + Query.renderErrorLocation(query, new ErrorLocation(e.getLineNumber(), e.getColumnNumber()), System.out); + } + catch (IOException e) { + e.printStackTrace(System.err); + } + } + } + + private void createIndexCommand(String query, QueryRunner queryRunner, AtomicBoolean exiting) + { + boolean hasAuthenticated = executeCommand( + queryRunner, + exiting, + "show catalogs;", + ClientOptions.OutputFormat.NULL, + false, + false); + + if (hasAuthenticated) { + SqlParser parser = new SqlParser(); + + try { + CreateIndex createIndex = (CreateIndex) parser.createStatement(query); + String[] columns = createIndex.getColumnAliases().stream() + .map(Identifier::getValue) + .toArray(String[]::new); + + if (columns.length > 1) { + System.out.println("Composite indices are currently not supported"); + return; + } + Expression expression = null; + + // Separating the properties needed by IndexCommand Class from the properties for creating the heuristic index + final String parallelCreation = "parallelCreation"; + final String verbose = "verbose"; + List indexCommandClassProperties = Arrays.asList(parallelCreation, verbose); + + Map> properties = createIndex.getProperties().stream() + .collect(Collectors.partitioningBy(property -> indexCommandClassProperties.stream().anyMatch(property.getName().getValue()::equalsIgnoreCase))); + + String[] indexProperties = properties.get(false).stream() + .map(property -> property.getName() + "=" + property.getValue()) + .toArray(String[]::new); + Map providedClassProperties = properties.get(true).stream() + .collect(Collectors.toMap(property -> property.getName().getValue().toLowerCase(ENGLISH), + property -> Boolean.parseBoolean(property.getValue().toString()))); + + if (createIndex.getExpression().isPresent()) { + expression = createIndex.getExpression().get(); + } + String[] partitions = extractPartitions(expression).toArray(new String[0]); + + String indexTypes = createIndex.getIndexType(); + + IndexCommand command = new IndexCommand(clientOptions.configDirPath, createIndex.getIndexName().toString(), createIndex.getTableName().toString(), + columns, partitions, indexTypes, indexProperties, providedClassProperties.getOrDefault(parallelCreation, false), + providedClassProperties.getOrDefault(verbose, false), clientOptions.user); + command.createIndex(); + } + catch (ParsingException e) { + System.out.println(e.getMessage()); + Query.renderErrorLocation(query, new ErrorLocation(e.getLineNumber(), e.getColumnNumber()), System.out); + } + } + } + + private List extractPartitions(Expression expression) + { + if (expression instanceof ComparisonExpression) { + ComparisonExpression exp = (ComparisonExpression) expression; + return Collections.singletonList(exp.getLeft().toString() + "=" + exp.getRight().toString()); + } + else if (expression instanceof LogicalBinaryExpression) { + LogicalBinaryExpression exp = (LogicalBinaryExpression) expression; + Expression left = exp.getLeft(); + Expression right = exp.getRight(); + return Stream.concat(extractPartitions(left).stream(), extractPartitions(right).stream()).collect(Collectors.toList()); + } + return Collections.emptyList(); + } + + private void runConsole(QueryRunner queryRunner, AtomicBoolean exiting) { try (TableNameCompleter tableNameCompleter = new TableNameCompleter(queryRunner); InputReader reader = new InputReader(getHistoryFile(), Completion.commandCompleter(), tableNameCompleter)) { @@ -247,6 +443,11 @@ public class Console if (split.terminator().equals("\\G")) { outputFormat = ClientOptions.OutputFormat.VERTICAL; } + if (createIndexPattern.matcher(split.statement()).matches()) { + String query = split.statement(); + executeHeuristicIndexQuery(query, exiting, queryRunner); + continue; + } process(queryRunner, split.statement(), outputFormat, tableNameCompleter::populateCache, true, true, reader.getTerminal(), System.out, System.out); } diff --git a/presto-cli/src/main/java/io/prestosql/cli/Query.java b/presto-cli/src/main/java/io/prestosql/cli/Query.java index 1afbd3ec0..cddd243fa 100644 --- a/presto-cli/src/main/java/io/prestosql/cli/Query.java +++ b/presto-cli/src/main/java/io/prestosql/cli/Query.java @@ -355,7 +355,7 @@ public class Query out.println(); } - private static void renderErrorLocation(String query, ErrorLocation location, PrintStream out) + static void renderErrorLocation(String query, ErrorLocation location, PrintStream out) { List lines = ImmutableList.copyOf(Splitter.on('\n').split(query).iterator()); diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java index f4175a646..1b2b09790 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java @@ -19,6 +19,7 @@ import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; import com.google.common.cache.Weigher; +import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.inject.Inject; import io.airlift.log.Logger; @@ -48,6 +49,7 @@ public class IndexCache { private static final Logger LOG = Logger.get(IndexCache.class); private static final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("Hive-IndexCache-pool-%d").setDaemon(true).build(); + private static final List INDEX_TYPES = ImmutableList.of("bloom", "minmax"); private static ScheduledExecutorService executor; @@ -103,24 +105,26 @@ public class IndexCache // for each split, load indexes for each predicate (if the predicate contains an indexed column) List splitIndexes = new LinkedList<>(); effectivePredicate.getDomains().get().keySet().stream() - // if the domain column is a partition column, skip it - .filter(key -> partitions == null || !partitions.contains(key)) - .map(HiveColumnHandle::getName) - .map(String::toLowerCase).forEach(column -> { - // security check required before using values in a Path - // e.g. catalog.schema.table or dc.catalog.schema.table - if (!tableFqn.matches("([\\p{Alnum}_]+\\.){2,3}[\\p{Alnum}_]+")) { - LOG.warn("Invalid table name " + tableFqn); - return; - } + // if the domain column is a partition column, skip it + .filter(key -> partitions == null || !partitions.contains(key)) + .map(HiveColumnHandle::getName) + .map(String::toLowerCase) + .forEach(column -> { + // security check required before using values in a Path + // e.g. catalog.schema.table or dc.catalog.schema.table + if (!tableFqn.matches("([\\p{Alnum}_]+\\.){2,3}[\\p{Alnum}_]+")) { + LOG.warn("Invalid table name " + tableFqn); + return; + } - if (!column.matches("[\\p{Alnum}_]+")) { - LOG.warn("Invalid column name " + column); - return; - } + if (!column.matches("[\\p{Alnum}_]+")) { + LOG.warn("Invalid column name " + column); + return; + } - String indexCacheKeyPath = Paths.get(tableFqn, column, pathUri.getPath()).toString(); - IndexCacheKey indexCacheKey = new IndexCacheKey(indexCacheKeyPath, lastModifiedTime, "bitmap", "bloom"); + for (String indexType : INDEX_TYPES) { + String indexCacheKeyPath = Paths.get(tableFqn, column, indexType, pathUri.getPath()).toString(); + IndexCacheKey indexCacheKey = new IndexCacheKey(indexCacheKeyPath, lastModifiedTime); // check if cache contains the key List predicateIndexes = cache.getIfPresent(indexCacheKey); @@ -154,7 +158,8 @@ public class IndexCache splitIndexes.addAll(predicateIndexes); } } - }); + } + }); return splitIndexes; } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCacheLoader.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCacheLoader.java index cc26d9053..a81f85af6 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCacheLoader.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCacheLoader.java @@ -20,7 +20,6 @@ import io.hetu.core.common.heuristicindex.IndexCacheKey; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexMetadata; -import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @@ -62,7 +61,7 @@ public class IndexCacheLoader List indices; try { - indices = indexClient.readSplitIndex(key.getPath(), key.getIndexTypes()); + indices = indexClient.readSplitIndex(key.getPath()); } catch (Exception e) { throw new Exception("No valid index files found for key " + key, e); @@ -70,7 +69,7 @@ public class IndexCacheLoader // lastModified file was valid, but no index files for the given types if (indices.isEmpty()) { - throw new Exception("No index files found for key " + key + " of type(s) " + Arrays.toString(key.getIndexTypes())); + throw new Exception("No index files found for key " + key); } // Sort the indices based on split starting position diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java index 9c9bb9847..ad8c9ac17 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java @@ -106,7 +106,7 @@ public class TestIndexCache assertEquals(actualSplitIndex.size(), 0); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicate, testPartitions); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); } @Test @@ -156,7 +156,7 @@ public class TestIndexCache Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicate, testPartitions); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); // now the index is in the cache, but changing the lastmodified date of the split should invalidate it when(testHiveSplit.getLastModifiedTime()).thenReturn(testLastModifiedTime + 1); @@ -190,7 +190,7 @@ public class TestIndexCache actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicateForPartition, partitionColumns); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); } @Test @@ -216,9 +216,9 @@ public class TestIndexCache Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicate, testPartitions); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); assertEquals(actualSplitIndex.get(0), expectedIndices1.get(0)); - assertEquals(indexCache.getCacheSize(), 1); + assertEquals(indexCache.getCacheSize(), 2); //get index for split2 when(testHiveSplit.getPath()).thenReturn(testPath2); @@ -233,7 +233,7 @@ public class TestIndexCache actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicate, testPartitions); assertEquals(actualSplitIndex.size(), 0); - assertEquals(indexCache.getCacheSize(), 1); + assertEquals(indexCache.getCacheSize(), 2); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit, effectivePredicate, testPartitions); assertEquals(actualSplitIndex.size(), 1); diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java index 004d32f23..a8772ddfe 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java @@ -19,6 +19,7 @@ import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; import com.google.common.cache.Weigher; +import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.ThreadFactoryBuilder; import io.airlift.log.Logger; import io.hetu.core.common.heuristicindex.IndexCacheKey; @@ -29,6 +30,7 @@ import io.prestosql.spi.service.PropertyService; import java.net.URI; import java.util.Collections; +import java.util.LinkedList; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; @@ -42,6 +44,7 @@ public class IndexCache { private static final Logger LOG = Logger.get(IndexCache.class); private static final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("Main-IndexCache-pool-%d").setDaemon(true).build(); + private static final List INDEX_TYPES = ImmutableList.of("bloom", "minmax"); private static ScheduledExecutorService executor; @@ -81,49 +84,49 @@ public class IndexCache } URI splitUri = URI.create(split.getConnectorSplit().getFilePath()); - String filterKeyPath = getCacheKey(table, column, splitUri.getPath()); long lastModifiedTime = split.getConnectorSplit().getLastModifiedTime(); - IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime, "MINMAX", "BLOOM"); - //it is possible to return multiple SplitIndexMetadata due to the range mismatch, especially in the case - //where the split has a wider range than the original splits used for index creation - // check if cache contains the key - List indices; + List indices = new LinkedList<>(); - // if cache didn't contain the key, it has not been loaded, load it asynchronously - indices = cache.getIfPresent(filterKey); + for (String indexType : INDEX_TYPES) { + String filterKeyPath = table + "/" + column + "/" + indexType + splitUri.getPath(); + IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime); + //it is possible to return multiple SplitIndexMetadata due to the range mismatch, especially in the case + //where the split has a wider range than the original splits used for index creation + // check if cache contains the key + List indexOfThisType; - if (indices == null) { - executor.schedule(() -> { - try { - cache.get(filterKey); - LOG.debug("Loaded index for %s.", filterKey); - } - catch (ExecutionException e) { - if (LOG.isDebugEnabled()) { - LOG.debug(e, "Unable to load index for %s. ", filterKey); + // if cache didn't contain the key, it has not been loaded, load it asynchronously + indexOfThisType = cache.getIfPresent(filterKey); + + if (indexOfThisType == null) { + executor.schedule(() -> { + try { + cache.get(filterKey); + LOG.debug("Loaded index for %s.", filterKey); + } + catch (ExecutionException e) { + if (LOG.isDebugEnabled()) { + LOG.debug(e, "Unable to load index for %s. ", filterKey); + } + } + }, loadDelay, TimeUnit.MILLISECONDS); + } + + if (indexOfThisType != null) { + // if key was present in cache, we still need to check if the index is validate based on the lastModifiedTime + // the index is only valid if the lastModifiedTime of the split matches the index's lastModifiedTime + for (IndexMetadata index : indexOfThisType) { + if (index.getLastUpdated() != lastModifiedTime) { + cache.invalidate(filterKey); + indexOfThisType = Collections.emptyList(); + break; } } - }, loadDelay, TimeUnit.MILLISECONDS); - } - - if (indices != null) { - // if key was present in cache, we still need to check if the index is validate based on the lastModifiedTime - // the index is only valid if the lastModifiedTime of the split matches the index's lastModifiedTime - for (IndexMetadata index : indices) { - if (index.getLastUpdated() != lastModifiedTime) { - cache.invalidate(filterKey); - indices = Collections.emptyList(); - break; - } + indices.addAll(indexOfThisType); } } - return indices == null ? Collections.emptyList() : indices; - } - - private String getCacheKey(String tableName, String columnName, String filePath) - { - return tableName + "/" + columnName + filePath; + return indices; } @VisibleForTesting diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java index ef35ce707..bc16658d1 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java @@ -19,7 +19,6 @@ import io.hetu.core.common.heuristicindex.IndexCacheKey; import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexMetadata; -import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @@ -60,7 +59,7 @@ public class IndexCacheLoader List indices; try { - indices = indexClient.readSplitIndex(key.getPath(), key.getIndexTypes()); + indices = indexClient.readSplitIndex(key.getPath()); } catch (Exception e) { throw new Exception("No valid index file found for key " + key, e); @@ -68,7 +67,7 @@ public class IndexCacheLoader // lastModified file was valid, but no index files for the given types if (indices.isEmpty()) { - throw new Exception("No index files found for key " + key + " of type(s) " + Arrays.toString(key.getIndexTypes())); + throw new Exception("No index files found for key " + key); } // Sort the indices based on split starting position diff --git a/presto-main/src/main/java/io/prestosql/testing/NoOpIndexClient.java b/presto-main/src/main/java/io/prestosql/testing/NoOpIndexClient.java index 99c475e22..db984620b 100644 --- a/presto-main/src/main/java/io/prestosql/testing/NoOpIndexClient.java +++ b/presto-main/src/main/java/io/prestosql/testing/NoOpIndexClient.java @@ -23,7 +23,7 @@ public class NoOpIndexClient implements IndexClient { @Override - public List readSplitIndex(String path, String... filterIndexTypes) + public List readSplitIndex(String path) throws IOException { throw new UnsupportedOperationException("This is a no-op index client"); @@ -37,7 +37,7 @@ public class NoOpIndexClient } @Override - public void deleteIndex(String table, String[] columns) + public void deleteIndex(String table, String[] columns, String indexType) throws IOException { throw new UnsupportedOperationException("This is a no-op index client"); diff --git a/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java b/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java index add56fddb..516e43c3b 100644 --- a/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java +++ b/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java @@ -90,7 +90,7 @@ public class TestIndexCache assertEquals(actualSplitIndex.size(), 0); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(table, column, split); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); assertEquals(actualSplitIndex.get(0), expectedIndices.get(0)); } @@ -136,7 +136,7 @@ public class TestIndexCache assertEquals(actualSplitIndex.size(), 0); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(table, column, split); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); // now the index is in the cache, but changing the lastmodified date of the split should invalidate it when(indexMetadata.getLastUpdated()).then(new Returns(testLastModifiedTime + 1)); @@ -166,9 +166,9 @@ public class TestIndexCache assertEquals(actualSplitIndex.size(), 0); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(table, column, split); - assertEquals(actualSplitIndex.size(), 1); + assertEquals(actualSplitIndex.size(), 2); assertEquals(actualSplitIndex.get(0), indexMetadata1); - assertEquals(indexCache.getCacheSize(), 1); + assertEquals(indexCache.getCacheSize(), 2); //get index for split2 when(connectorSplit.getFilePath()).thenReturn(testPath2); @@ -183,7 +183,7 @@ public class TestIndexCache when(indexCacheLoader.load(any())).then(new Returns(expectedIndices2)); actualSplitIndex = indexCache.getIndices(table, column, split); assertEquals(actualSplitIndex.size(), 0); - assertEquals(indexCache.getCacheSize(), 1); + assertEquals(indexCache.getCacheSize(), 2); Thread.sleep(loadDelay + 500); actualSplitIndex = indexCache.getIndices(table, column, split); assertEquals(actualSplitIndex.size(), 1); 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 6da5714ed..81d135e58 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 @@ -54,6 +54,15 @@ statement | DROP CACHE (IF EXISTS)? qualifiedName (WHERE booleanExpression)? #dropCache | SHOW CACHE qualifiedName? #showCache + | CREATE INDEX (IF NOT EXISTS)? indexName=qualifiedName + USING indexType + ON tableName=qualifiedName columnAliases + (WITH properties)? + (WHERE expression)? #createIndex + | DROP INDEX (IF EXISTS)? indexName=qualifiedName #dropIndex + | ALTER INDEX (IF EXISTS)? from=qualifiedName RENAME TO to=qualifiedName #renameIndex + | UPDATE INDEX (IF EXISTS)? qualifiedName (SET properties)? #updateIndex + | SHOW INDEX (IF EXISTS)? qualifiedName? #showIndex | INSERT INTO qualifiedName columnAliases? query #insertInto | INSERT OVERWRITE (TABLE)? qualifiedName columnAliases? query #insertOverwrite | DELETE FROM qualifiedName (WHERE booleanExpression)? #delete @@ -373,6 +382,10 @@ booleanValue : TRUE | FALSE ; +indexType + : BITMAP | BLOOM | MINMAX + ; + interval : INTERVAL sign=(PLUS | MINUS)? string from=intervalField (TO to=intervalField)? ; @@ -531,6 +544,7 @@ nonReserved | WORK | WRITE | YEAR | ZONE + | INDEX ; ADD: 'ADD'; @@ -734,6 +748,10 @@ WORK: 'WORK'; WRITE: 'WRITE'; YEAR: 'YEAR'; ZONE: 'ZONE'; +INDEX: 'INDEX'; +BITMAP: 'BITMAP'; +BLOOM: 'BLOOM'; +MINMAX: 'MINMAX'; EQ : '='; NEQ : '<>' | '!='; 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 5053831ff..e7561cc1c 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java +++ b/presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java @@ -27,6 +27,7 @@ import io.prestosql.sql.tree.CallArgument; import io.prestosql.sql.tree.ColumnDefinition; import io.prestosql.sql.tree.Comment; import io.prestosql.sql.tree.Commit; +import io.prestosql.sql.tree.CreateIndex; import io.prestosql.sql.tree.CreateRole; import io.prestosql.sql.tree.CreateSchema; import io.prestosql.sql.tree.CreateTable; @@ -38,6 +39,7 @@ import io.prestosql.sql.tree.DescribeInput; import io.prestosql.sql.tree.DescribeOutput; import io.prestosql.sql.tree.DropCache; import io.prestosql.sql.tree.DropColumn; +import io.prestosql.sql.tree.DropIndex; import io.prestosql.sql.tree.DropRole; import io.prestosql.sql.tree.DropSchema; import io.prestosql.sql.tree.DropTable; @@ -95,6 +97,7 @@ import io.prestosql.sql.tree.ShowColumns; import io.prestosql.sql.tree.ShowCreate; import io.prestosql.sql.tree.ShowFunctions; import io.prestosql.sql.tree.ShowGrants; +import io.prestosql.sql.tree.ShowIndex; import io.prestosql.sql.tree.ShowRoleGrants; import io.prestosql.sql.tree.ShowRoles; import io.prestosql.sql.tree.ShowSchemas; @@ -835,6 +838,47 @@ public final class SqlFormatter return null; } + @Override + protected Void visitCreateIndex(CreateIndex node, Integer context) + { + builder.append("CREATE INDEX "); + if (node.isNotExists()) { + builder.append("IF NOT EXISTS "); + } + builder.append(formatName(node.getIndexName())); + builder.append("USING "); + builder.append(format(node.getIndexType())); + builder.append(" ON "); + builder.append(formatName(node.getTableName())); + String columnList = node.getColumnAliases().stream().map(element -> formatExpression(element, parameters)).collect(joining(", ")); + builder.append(format("( %s )", columnList)); + builder.append(formatPropertiesMultiLine(node.getProperties())); + if (node.getExpression().isPresent()) { + builder.append(node.getExpression().get()); + } + + return null; + } + + @Override + protected Void visitDropIndex(DropIndex node, Integer context) + { + append(context, "DROP INDEX "); + if (node.isExists()) { + builder.append("IF EXISTS "); + } + builder.append(node.getIndexName()); + + return null; + } + + @Override + protected Void visitShowIndex(ShowIndex node, Integer context) + { + append(context, "SHOW INDEX "); + return null; + } + @Override protected Void visitCreateTableAsSelect(CreateTableAsSelect 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 6baec623e..97465d0f1 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 @@ -40,6 +40,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.CreateIndex; import io.prestosql.sql.tree.CreateRole; import io.prestosql.sql.tree.CreateSchema; import io.prestosql.sql.tree.CreateTable; @@ -58,6 +59,7 @@ import io.prestosql.sql.tree.DescribeOutput; import io.prestosql.sql.tree.DoubleLiteral; import io.prestosql.sql.tree.DropCache; import io.prestosql.sql.tree.DropColumn; +import io.prestosql.sql.tree.DropIndex; import io.prestosql.sql.tree.DropRole; import io.prestosql.sql.tree.DropSchema; import io.prestosql.sql.tree.DropTable; @@ -147,6 +149,7 @@ import io.prestosql.sql.tree.ShowColumns; import io.prestosql.sql.tree.ShowCreate; import io.prestosql.sql.tree.ShowFunctions; import io.prestosql.sql.tree.ShowGrants; +import io.prestosql.sql.tree.ShowIndex; import io.prestosql.sql.tree.ShowRoleGrants; import io.prestosql.sql.tree.ShowRoles; import io.prestosql.sql.tree.ShowSchemas; @@ -362,6 +365,49 @@ class AstBuilder return new DropView(getLocation(context), getQualifiedName(context.qualifiedName()), context.EXISTS() != null); } + @Override + public Node visitCreateIndex(SqlBaseParser.CreateIndexContext context) + { + List properties = ImmutableList.of(); + if (context.properties() != null) { + properties = visit(context.properties().property(), Property.class); + } + List columnAliases = ImmutableList.of(); + if (context.columnAliases() != null) { + columnAliases = visit(context.columnAliases().identifier(), Identifier.class); + } + Optional indexType = Optional.empty(); + if (context.indexType() != null) { + indexType = Optional.of(visitIfPresent(context.indexType(), StringLiteral.class).get().getValue()); + } + return new CreateIndex( + getLocation(context), + getQualifiedName(context.indexName), + getQualifiedName(context.tableName), + columnAliases, + indexType.orElse(null), + context.EXISTS() != null, + properties, + visitIfPresent(context.expression(), Expression.class)); + } + + @Override + public Node visitDropIndex(SqlBaseParser.DropIndexContext context) + { + return new DropIndex(getLocation(context), getQualifiedName(context.qualifiedName()), context.EXISTS() != null); + } + + @Override + public Node visitShowIndex(SqlBaseParser.ShowIndexContext context) + { + if (context.qualifiedName() != null) { + return new ShowIndex(getLocation(context), getQualifiedName(context.qualifiedName())); + } + else { + return new ShowIndex(getLocation(context)); + } + } + @Override public Node visitInsertInto(SqlBaseParser.InsertIntoContext context) { @@ -1866,6 +1912,12 @@ class AstBuilder return new BooleanLiteral(getLocation(context), context.getText()); } + @Override + public Node visitIndexType(SqlBaseParser.IndexTypeContext context) + { + return new StringLiteral(getLocation(context), context.getText()); + } + @Override public Node visitInterval(SqlBaseParser.IntervalContext context) { diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/AstVisitor.java b/presto-parser/src/main/java/io/prestosql/sql/tree/AstVisitor.java index 17782e973..ec71374e7 100644 --- a/presto-parser/src/main/java/io/prestosql/sql/tree/AstVisitor.java +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/AstVisitor.java @@ -582,6 +582,21 @@ public abstract class AstVisitor return visitStatement(node, context); } + protected R visitCreateIndex(CreateIndex node, C context) + { + return visitStatement(node, context); + } + + protected R visitDropIndex(DropIndex node, C context) + { + return visitStatement(node, context); + } + + protected R visitShowIndex(ShowIndex node, C context) + { + return visitStatement(node, context); + } + protected R visitComment(Comment node, C context) { return visitStatement(node, context); diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/CreateIndex.java b/presto-parser/src/main/java/io/prestosql/sql/tree/CreateIndex.java new file mode 100644 index 000000000..8924b72cc --- /dev/null +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/CreateIndex.java @@ -0,0 +1,150 @@ +/* + * 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. + */ +package io.prestosql.sql.tree; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +import static com.google.common.base.MoreObjects.toStringHelper; +import static java.util.Objects.requireNonNull; + +public class CreateIndex + extends Statement +{ + private final QualifiedName indexName; + private final QualifiedName tableName; + private final List columnAliases; + private final String indexType; + private final boolean notExists; + private final List properties; + private final Optional where; + + public CreateIndex(QualifiedName indexName, QualifiedName tableName, List columnAliases, + String indexType, boolean notExists, List properties) + { + this(Optional.empty(), indexName, tableName, columnAliases, indexType, notExists, properties, Optional.empty()); + } + + public CreateIndex(NodeLocation location, QualifiedName indexName, QualifiedName tableName, List columnAliases, + String indexType, boolean notExists, List properties, Optional where) + { + this(Optional.of(location), indexName, tableName, columnAliases, indexType, notExists, properties, where); + } + + private CreateIndex(Optional location, QualifiedName indexName, QualifiedName tableName, List columnAliases, + String indexType, boolean notExists, List properties, Optional where) + { + super(location); + this.indexName = requireNonNull(indexName, "indexName is null"); + this.tableName = requireNonNull(tableName, "tableName is null"); + this.columnAliases = columnAliases; + this.indexType = requireNonNull(indexType, "indexType is null"); + this.notExists = notExists; + this.properties = ImmutableList.copyOf(requireNonNull(properties, "properties is null")); + this.where = where; + } + + public QualifiedName getIndexName() + { + return indexName; + } + + public QualifiedName getTableName() + { + return tableName; + } + + public List getColumnAliases() + { + return columnAliases; + } + + public String getIndexType() + { + return indexType; + } + + public boolean isNotExists() + { + return notExists; + } + + public List getProperties() + { + return properties; + } + + public Optional getExpression() + { + return where; + } + + @Override + public R accept(AstVisitor visitor, C context) + { + return visitor.visitCreateIndex(this, context); + } + + @Override + public List getChildren() + { + return ImmutableList.builder() + .addAll(columnAliases) + .addAll(properties) + .build(); + } + + @Override + public int hashCode() + { + return Objects.hash(indexName, tableName, columnAliases, indexType, notExists, properties, where); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if ((obj == null) || (getClass() != obj.getClass())) { + return false; + } + CreateIndex o = (CreateIndex) obj; + return Objects.equals(indexName, o.indexName) + && Objects.equals(tableName, o.tableName) + && Objects.equals(columnAliases, o.columnAliases) + && Objects.equals(indexType, o.indexType) + && Objects.equals(notExists, o.notExists) + && Objects.equals(properties, o.properties) + && Objects.equals(where, o.where); + } + + @Override + public String toString() + { + return toStringHelper(this) + .add("indexName", indexName) + .add("tableName", tableName) + .add("columnAliases", columnAliases) + .add("indexType", indexType) + .add("notExists", notExists) + .add("properties", properties) + .add("where", where) + .toString(); + } +} diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/DropIndex.java b/presto-parser/src/main/java/io/prestosql/sql/tree/DropIndex.java new file mode 100644 index 000000000..63d1a641c --- /dev/null +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/DropIndex.java @@ -0,0 +1,97 @@ +/* + * 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.sql.tree; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +import static com.google.common.base.MoreObjects.toStringHelper; + +public class DropIndex + extends Statement +{ + private final QualifiedName indexName; + private final boolean exists; + + public DropIndex(QualifiedName indexName, boolean exists) + { + this(Optional.empty(), indexName, exists); + } + + public DropIndex(NodeLocation location, QualifiedName indexName, boolean exists) + { + this(Optional.of(location), indexName, exists); + } + + private DropIndex(Optional location, QualifiedName indexName, boolean exists) + { + super(location); + this.indexName = indexName; + this.exists = exists; + } + + public QualifiedName getIndexName() + { + return indexName; + } + + public boolean isExists() + { + return exists; + } + + @Override + public R accept(AstVisitor visitor, C context) + { + return visitor.visitDropIndex(this, context); + } + + @Override + public List getChildren() + { + return ImmutableList.of(); + } + + @Override + public int hashCode() + { + return Objects.hash(indexName, exists); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if ((obj == null) || (getClass() != obj.getClass())) { + return false; + } + DropIndex o = (DropIndex) obj; + return Objects.equals(indexName, o.indexName) + && (exists == o.exists); + } + + @Override + public String toString() + { + return toStringHelper(this) + .add("indexName", indexName) + .add("exists", exists) + .toString(); + } +} diff --git a/presto-parser/src/main/java/io/prestosql/sql/tree/ShowIndex.java b/presto-parser/src/main/java/io/prestosql/sql/tree/ShowIndex.java new file mode 100644 index 000000000..361bdbe70 --- /dev/null +++ b/presto-parser/src/main/java/io/prestosql/sql/tree/ShowIndex.java @@ -0,0 +1,82 @@ +/* + * 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. + */ +package io.prestosql.sql.tree; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Optional; + +import static com.google.common.base.MoreObjects.toStringHelper; + +public class ShowIndex + extends Statement +{ + private final Optional indexName; + + public ShowIndex(NodeLocation location, QualifiedName indexName) + { + this(Optional.of(location), Optional.of(indexName)); + } + + public ShowIndex(NodeLocation location) + { + this(Optional.of(location), Optional.empty()); + } + + public ShowIndex(Optional location, Optional indexName) + { + super(location); + this.indexName = indexName; + } + + public QualifiedName getIndexName() + { + return indexName.isPresent() ? indexName.get() : null; + } + + @Override + public R accept(AstVisitor visitor, C context) + { + return visitor.visitShowIndex(this, context); + } + + @Override + public List getChildren() + { + return ImmutableList.of(); + } + + @Override + public int hashCode() + { + return 0; + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + return (obj != null) && (getClass() == obj.getClass()); + } + + @Override + public String toString() + { + return toStringHelper(this).toString(); + } +} diff --git a/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParserErrorHandling.java b/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParserErrorHandling.java index accff0b86..b16aba555 100644 --- a/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParserErrorHandling.java +++ b/presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParserErrorHandling.java @@ -88,7 +88,7 @@ public class TestSqlParserErrorHandling {"select foo(DISTINCT ,1)", "line 1:21: mismatched input ','. Expecting: "}, {"CREATE )", - "line 1:8: mismatched input ')'. Expecting: 'DATABASE', 'OR', 'ROLE', 'SCHEMA', 'TABLE', 'VIEW'"}, + "line 1:8: mismatched input ')'. Expecting: 'DATABASE', 'INDEX', 'OR', 'ROLE', 'SCHEMA', 'TABLE', 'VIEW'"}, {"CREATE TABLE ) AS (VALUES 1)", "line 1:14: mismatched input ')'. Expecting: 'IF', "}, {"CREATE TABLE foo ", diff --git a/presto-spi/src/main/java/io/prestosql/spi/connector/CreateIndexMetadata.java b/presto-spi/src/main/java/io/prestosql/spi/connector/CreateIndexMetadata.java new file mode 100644 index 000000000..cd859b7a5 --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/connector/CreateIndexMetadata.java @@ -0,0 +1,134 @@ +/* + * 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.connector; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; + +import static io.prestosql.spi.connector.SchemaUtil.checkNotEmpty; +import static java.util.Objects.requireNonNull; + +public class CreateIndexMetadata +{ + private final String indexName; + private final String tableName; + private final String indexType; + private final List indexColumns; + private final String expression; + private final List properties; + + @JsonCreator + public CreateIndexMetadata( + @JsonProperty("indexName") String indexName, + @JsonProperty("tableName") String tableName, + @JsonProperty("indexType") String indexType, + @JsonProperty("indexColumns") List indexColumns, + @JsonProperty("expression") String expression, + @JsonProperty("properties") List properties) + { + checkNotEmpty(indexName, "indexName"); + this.indexName = indexName; + requireNonNull(tableName, "tableName is null"); + this.tableName = tableName; + requireNonNull(indexType, "indexType is null"); + this.indexType = indexType; + this.indexColumns = indexColumns; + this.expression = expression; + this.properties = ImmutableList.copyOf(requireNonNull(properties, "properties is null")); + } + + @JsonProperty + public String getIndexName() + { + return indexName; + } + + @JsonProperty + public String getTableName() + { + return tableName; + } + + @JsonProperty + public String getIndexType() + { + return indexType; + } + + @JsonProperty + public List getIndexColumns() + { + return indexColumns; + } + + @JsonProperty + public String getExpression() + { + return expression; + } + + @JsonProperty + public List getProperties() + { + return properties; + } + + @Override + public String toString() + { + StringBuilder sb = new StringBuilder("CreateIndexMetadata{"); + sb.append("indexName='").append(indexName).append('\''); + sb.append("tableName='").append(tableName).append('\''); + sb.append("indexType=").append(indexType).append('\''); + if (!indexColumns.isEmpty()) { + sb.append(", indexColumns=").append(indexColumns); + } + if (expression != null) { + sb.append(", expression='").append(expression).append('\''); + } + if (!properties.isEmpty()) { + sb.append(", properties=").append(properties); + } + sb.append('}'); + return sb.toString(); + } + + @Override + public int hashCode() + { + return Objects.hash(indexName, tableName, indexType, indexColumns, expression, properties); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + CreateIndexMetadata other = (CreateIndexMetadata) obj; + return Objects.equals(this.indexName, other.indexName) && + Objects.equals(this.tableName, other.tableName) && + Objects.equals(this.indexType, other.indexType) && + Objects.equals(this.indexColumns, other.indexColumns) && + Objects.equals(this.expression, other.expression) && + Objects.equals(this.properties, other.properties); + } +} diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/DataSource.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/DataSource.java index 13f95e8eb..b7e0d9d2f 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/DataSource.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/DataSource.java @@ -111,7 +111,8 @@ public interface DataSource * @param uri uri of the file or source that was read * @param splitStart the split offset, e.g. if the source was a large file, * it may have been read in multiple splits + * @param progress Value between 0.0 and 1.0 representing the progress. */ - void call(String column, Object[] values, String uri, long splitStart, long lastModified); + void call(String column, Object[] values, String uri, long splitStart, long lastModified, double progress); } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexClient.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexClient.java index 8bb353d2c..60414179b 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexClient.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexClient.java @@ -52,11 +52,11 @@ public interface IndexClient * * @param path relative path to the split index file or dir, if dir, it will be searched recursively (relative to \ * the root uri, if one was set) - * @param filterIndexTypes only load index types matching these types, if empty or null, all types will be loaded * @return all split indexes that were read, with the split metadata set based on the split path * @throws IOException thrown by doing IO operations using filesystem client */ - public List readSplitIndex(String path, String... filterIndexTypes) throws IOException; + public List readSplitIndex(String path) + throws IOException; /** * Searches the path for lastModified file and returns the value as a long. @@ -77,10 +77,11 @@ public interface IndexClient * Delete the indexes for the table, if columns are specified, only indexes * for the specified columns will be deleted. * - * @param table fully qualified table name - * @param columns columns to delete index of, if null the indexes for the entire - * table are deleted + * @param table table of the index + * @param columns columns of the index + * @param indexType the index type * @throws IOException any IOException thrown by filesystem client during file deletion */ - public void deleteIndex(String table, String[] columns) throws IOException; + public void deleteIndex(String table, String[] columns, String indexType) + throws IOException; } diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexWriter.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexWriter.java index 2c3047ef9..25247e29c 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexWriter.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexWriter.java @@ -51,14 +51,13 @@ public interface IndexWriter * @param table fully qualified table name * @param columns columns to index * @param partitions only index specified partitions, if null, index all partitions - * @param indexTypes type of the index to be created (its string ID returned from {@link Index#getId()}) + * @param indexType type of the index to be created (its string ID returned from {@link Index#getId()}) * @param lockingEnabled if enabled, the table will be locked and multiple callers can't create index for the table in parallel - * @param debugEnabled writes the raw split data to a file alongside the index file * @throws IOException thrown during index creation */ - public void createIndex(String table, String[] columns, String[] partitions, String[] indexTypes, boolean lockingEnabled, boolean debugEnabled) + public void createIndex(String table, String[] columns, String[] partitions, String indexType, boolean lockingEnabled) throws IOException; - public void createIndex(String table, String[] columns, String[] partitions, String... indexTypes) + public void createIndex(String table, String[] columns, String[] partitions, String indexType) throws IOException; }