jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!246 Integrate heuristic index cli into presto-cli 

Merge pull request !246 from AyushGupta3/index-sql-squashed
This commit is contained in:
i-robot 2020-09-22 15:51:43 +08:00 committed by Gitee
parent e40b8033c3 e9d4b75cf7
commit aa9a9cf486
53 changed files with 1801 additions and 960 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
hetu-common/src/main/java/io/hetu/core/common/heuristicindex/IndexCacheKey.java
View File

@ -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;
}
}

69
hetu-docs/en/indexer/indexer-cli.md
View File

@ -3,48 +3,26 @@
## Usage
The index executable will be located under the `bin` directory in the installation.
For example, `<path to installtion directory>/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=<table>
[-c=<configDirPath>] [--column=<columns>[,<columns>...]]...
[--partition=<partitions>[,<partitions>...]]...
[--type=<indexTypes>[,<indexTypes>...]]...
[-I=<indexproperties>[,<indexproperties>...]]<command>
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> command types, e.g. create, delete, show; Note: delete command
works a column level only.
--column=<columns>[,<columns>...]
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=<partitions>[,<partitions>...]
only create index for these partitions, comma separated
format for multiple partitions
--table=<table> fully qualified table name
--type=<indexTypes>[,<indexTypes>...]
index type, comma separated format for multiple types
(supported types: BLOOM, BITMAP, MINMAX
-c, --config=<configDirPath>
root folder of openLooKeng etc directory (default: ../etc)
-p, --plugins=<plugins>[,<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'
```

2
hetu-docs/en/indexer/overview.md
View File

@ -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 <your-etc-folder-directory> --table table1 --column id --type bloom create
./hetu-cli --config <your-etc-folder-directory> --execute 'CREATE INDEX index_name USING bloom ON table1 (column)'
### Run query

65
hetu-docs/zh/indexer/indexer-cli.md
View File

@ -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=<table>
[-c=<configDirPath>] [--column=<columns>[,<columns>...]]...
[--partition=<partitions>[,<partitions>...]]...
[--type=<indexTypes>[,<indexTypes>...]]...
[-I=<indexproperties>[,<indexproperties>...]]<command>
使用此索引工具，您可以创建、显示和删除索引。
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配置
<command> 命令类型，如create、delete、show等；说明：delete命令只作用于列级。
--column=<columns>[,<columns>...]
列，使用逗号分隔多个列
--debug 如果启用，则每个Split的原始数据也将随索引一起写入文件
--disableLocking 默认锁定在表级别启用；如果设置为false，用户必须确保相同的数据没有被多个调用方同时索引（允许不同列或分区并行索引）
--partition=<partitions>[,<partitions>...]
只为这些分区创建索引，用逗号分隔多个分区
--table=<table> 全量表名
--type=<indexTypes>[,<indexTypes>...]
索引类型，用逗号分隔多种类型 (支持的类型：BLOOM, BITMAP, MINMAX
-c, --config=<configDirPath>
openLooKeng etc目录的根目录（默认为../etc）
-p, --plugins=<plugins>[,<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'
```

2
hetu-docs/zh/indexer/overview.md
View File

@ -77,7 +77,7 @@
要创建索引，首先将工作目录cd至安装目录的`bin`文件夹，然后运行：
./index -c <your-etc-folder-directory> --table table1 --column id --type bloom create
./hetu-cli --config <your-etc-folder-directory> --execute 'CREATE INDEX index_name USING bloom ON table1 (column)'
### 运行语句

175
hetu-heuristic-index-cli/pom.xml
View File

@ -1,175 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>io.hetu.core</groupId>
<artifactId>presto-root</artifactId>
<version>316</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hetu-heuristic-index-cli</artifactId>
<version>${dep.hetu.version}</version>
<description>Indexer CLI module</description>
<properties>
<air.main.basedir>${project.parent.basedir}</air.main.basedir>
<air.check.fail-dependency>false</air.check.fail-dependency>
<air.check.fail-duplicate-finder>false</air.check.fail-duplicate-finder>
<dep.powermock.version>2.0.2</dep.powermock.version>
<org.slf4j.version>1.7.30</org.slf4j.version>
</properties>
<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${org.slf4j.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>info.picocli</groupId>
<artifactId>picocli</artifactId>
<version>4.0.0-alpha-3</version>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>2.28.2</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>net.bytebuddy</groupId>
<artifactId>byte-buddy-agent</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>hetu-heuristic-index</artifactId>
<exclusions>
<exclusion>
<groupId>org.checkerframework</groupId>
<artifactId>checker-qual</artifactId>
</exclusion>
<exclusion>
<groupId>javax.activation</groupId>
<artifactId>javax.activation-api</artifactId>
</exclusion>
<exclusion>
<groupId>javax.el</groupId>
<artifactId>javax.el-api</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>hetu-common</artifactId>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng</artifactId>
<version>${dep.powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito2</artifactId>
<version>${dep.powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng-common</artifactId>
<version>${dep.powermock.version}</version>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-core</artifactId>
<version>${dep.powermock.version}</version>
<exclusions>
</exclusions>
</dependency>
<dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>presto-spi</artifactId>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>hetu-filesystem-client</artifactId>
</dependency>
<dependency>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
<version>3.0.1</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
<version>3.24.0-GA</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>hetu-heuristic-index-cli-shaded</finalName>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" >
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>io.hetu.core.heuristicindex.IndexCommand</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<id>bin-tgz</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
<configuration>
<attach>true</attach>
<descriptors>
<descriptor>src/main/assemblies/bin.xml</descriptor>
</descriptors>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

14
hetu-heuristic-index-cli/src/main/bin/index
View File

@ -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\
"$@"

236
hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java
View File

@ -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<Void>
{
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<IndexMetadata> 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
}
}

37
hetu-heuristic-index/pom.xml
View File

@ -214,6 +214,43 @@
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>2.28.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-core</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-testng-common</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito2</artifactId>
<version>2.0.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
<version>3.0.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
<version>3.26.0-GA</version>
<scope>test</scope>
</dependency>
<dependency>

74
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java
View File

@ -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<IndexMetadata> readSplitIndex(String path, String... filterIndexTypes)
public List<IndexMetadata> readSplitIndex(String path)
throws IOException
{
requireNonNull(path, "no path specified");
List<IndexMetadata> indexes = new LinkedList<>();
for (Map.Entry<String, Index> entry : readIndexMap(path, filterIndexTypes).entrySet()) {
for (Map.Entry<String, Index> 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<String, Index> readIndexMap(String path, String... filterIndexTypes)
private Map<String, Index> readIndexMap(String path)
throws IOException
{
ImmutableMap.Builder<String, Index> result = ImmutableMap.builder();
@ -224,13 +214,13 @@ public class HeuristicIndexClient
try (Stream<Path> tarsOnRemote = fs.walk(absolutePath).filter(p -> p.toString().contains(".tar"))) {
for (Path tarFile : (Iterable<Path>) 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<Path> children = LOCAL_FS_CLIENT.walk(tarReadPath).filter(this::notDirectory)) {
for (Path child : (Iterable<Path>) 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) {

16
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java
View File

@ -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<Index> supportedIndex = ImmutableSet.of(new BloomIndex(), new MinMaxIndex(), new BitMapIndex());
private final Set<DataSource> 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<Index> 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);
}

172
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexWriter.java
View File

@ -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<String> 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<? extends Index> 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<? extends Index> 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<Path> tarsOnRemote = fs.walk(originalOnTarget).filter(p -> p.toString().contains(".tar"))) {
for (Path tarFile : (Iterable<Path>) 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<Path>) 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<String> 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<Path> children = LOCAL_FS_CLIENT.list(path).filter(this::notDirectory)) {
for (Path child : (Iterable<Path>) 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;
}
}

205
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexCommand.java Normal file
View File

@ -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<IndexRecordManager.IndexRecord> getIndexes()
{
try {
validatePaths();
IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath);
if (indexName.equals("")) {
return IndexRecordManager.readAllIndexRecords(indexStore.getFs(), indexStore.getRoot());
}
else {
List<IndexRecordManager.IndexRecord> 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());
}
}

246
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java Normal file
View File

@ -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<IndexRecord> readAllIndexRecords(HetuFileSystemClient fs, Path root)
throws IOException
{
validatePath(root);
Path recordFile = root.resolve(RECORD_FILE_NAME);
List<IndexRecord> 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<IndexRecord> 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<IndexRecord> 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<IndexRecord> records = readAllIndexRecords(fs, root);
String noteToWrite = String.join(",", note);
Iterator<IndexRecord> 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<IndexRecord> 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<IndexRecord> 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;
}
}
}

9
hetu-heuristic-index-cli/src/main/java/io/hetu/core/heuristicindex/IndexCommandUtils.java → hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java
View File

@ -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;

2
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexConstants.java
View File

@ -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";

18
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexServiceUtils.java
View File

@ -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);
}
}
}

11
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/datasource/hive/HdfsOrcDataSource.java
View File

@ -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<FileStatus> 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<Runnable>(10240));
try {
List<Future> 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<Integer, Type> columnTypes,
Map<Integer, String> 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();
}
}

20
hetu-heuristic-index/src/main/resources/log4j.properties
View File

@ -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

2
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/CsvDataSource.java
View File

@ -77,7 +77,7 @@ public class CsvDataSource
List<Object> 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) {

103
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexClient.java
View File

@ -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<TempFolder> 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<Index> 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<Index> emptyIndices = new HashSet<>();
HeuristicIndexClient client = new HeuristicIndexClient(emptyIndices, fs, folder.getRoot().toPath());
client.deleteIndex(tableName, new String[] {"testColumn"}, "bloom");
assertFalse(indexTypeFolder.exists());
}
}
}

13
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexFactory.java
View File

@ -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<IndexMetadata> 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;

25
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHeuristicIndexWriter.java
View File

@ -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<Path> 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);
}
}

83
hetu-heuristic-index-cli/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java → hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexCommand.java
View File

@ -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);
}
}

116
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java Normal file
View File

@ -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));
}
}
}

6
hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/datasource/hive/TestHiveDataSource.java
View File

@ -166,7 +166,7 @@ public class TestHiveDataSource
Map<String, Class> 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<String, Class> 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();
});

6
hetu-server/src/main/provisio/hetu.xml
View File

@ -12,8 +12,8 @@
</fileSet>
<fileSet to="bin">
<directory path="${basedir}/../hetu-heuristic-index-cli/target/">
<include>hetu-heuristic-index-cli-shaded.jar</include>
<directory path="${basedir}/../presto-cli/target/">
<include>hetu-cli-010-executable.jar</include>
</directory>
</fileSet>
@ -22,7 +22,7 @@
<artifact id="io.airlift:launcher:tar.gz:bin:${dep.packaging.version}">
<unpack />
</artifact>
<artifact id="io.hetu.core:hetu-heuristic-index-cli:tar.gz:bin:${project.version}">
<artifact id="io.hetu.core:presto-cli:tar.gz:bin:${project.version}">
<unpack />
</artifact>
<artifact id="io.airlift:launcher:tar.gz:properties:${dep.packaging.version}">

1
pom.xml
View File

@ -126,7 +126,6 @@
<module>hetu-oracle</module>
<module>hetu-vdm</module>
<module>hetu-heuristic-index</module>
<module>hetu-heuristic-index-cli</module>
<module>hetu-datacenter</module>
<module>hetu-hana</module>
<module>hetu-listener</module>

43
presto-cli/pom.xml
View File

@ -33,7 +33,7 @@
<groupId>org.antlr</groupId>
<artifactId>antlr4-runtime</artifactId>
</dependency>
<dependency>
<groupId>io.airlift</groupId>
<artifactId>airline</artifactId>
@ -118,6 +118,29 @@
<artifactId>jackson-core</artifactId>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>hetu-heuristic-index</artifactId>
<exclusions>
<exclusion>
<groupId>javax.el</groupId>
<artifactId>javax.el-api</artifactId>
</exclusion>
<exclusion>
<groupId>org.glassfish</groupId>
<artifactId>javax.el</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
<exclusion>
<groupId>org.checkerframework</groupId>
<artifactId>checker-qual</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- for testing -->
<dependency>
<groupId>org.testng</groupId>
@ -148,6 +171,7 @@
<shadedArtifactAttached>true</shadedArtifactAttached>
<shadedClassifierName>executable</shadedClassifierName>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<manifestEntries>
<Main-Class>${main-class}</Main-Class>
@ -158,20 +182,21 @@
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.skife.maven</groupId>
<artifactId>really-executable-jar-maven-plugin</artifactId>
<configuration>
<flags>-Xmx1G</flags>
<classifier>executable</classifier>
</configuration>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<id>bin-tgz</id>
<phase>package</phase>
<goals>
<goal>really-executable-jar</goal>
<goal>single</goal>
</goals>
<configuration>
<attach>true</attach>
<descriptors>
<descriptor>src/main/assemblies/bin.xml</descriptor>
</descriptors>
</configuration>
</execution>
</executions>
</plugin>

0
hetu-heuristic-index-cli/src/main/assemblies/bin.xml → presto-cli/src/main/assemblies/bin.xml
View File

4
presto-cli/src/main/bin/hetu-cli Executable file
View File

@ -0,0 +1,4 @@
#!/bin/bash
java -jar "$(dirname "$0")/hetu-cli-010-executable.jar"\
"$@"

3
presto-cli/src/main/java/io/prestosql/cli/ClientOptions.java
View File

@ -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,

205
presto-cli/src/main/java/io/prestosql/cli/Console.java
View File

@ -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<Column> columns = ImmutableList.<Column>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<IndexRecordManager.IndexRecord> records = command.getIndexes();
List<List<?>> rows = new ArrayList<>();
for (IndexRecordManager.IndexRecord v : records) {
List<String> 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<String> indexCommandClassProperties = Arrays.asList(parallelCreation, verbose);
Map<Boolean, List<Property>> 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<String, Boolean> 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<String> 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);
}

2
presto-cli/src/main/java/io/prestosql/cli/Query.java
View File

@ -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<String> lines = ImmutableList.copyOf(Splitter.on('\n').split(query).iterator());

39
presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java
View File

@ -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<String> 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<IndexMetadata> 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<IndexMetadata> predicateIndexes = cache.getIfPresent(indexCacheKey);
@ -154,7 +158,8 @@ public class IndexCache
splitIndexes.addAll(predicateIndexes);
}
}
});
}
});
return splitIndexes;
}

5
presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCacheLoader.java
View File

@ -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<IndexMetadata> 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

12
presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java
View File

@ -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);

73
presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java
View File

@ -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<String> 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<IndexMetadata> indices;
List<IndexMetadata> 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<IndexMetadata> 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

5
presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java
View File

@ -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<IndexMetadata> 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

4
presto-main/src/main/java/io/prestosql/testing/NoOpIndexClient.java
View File

@ -23,7 +23,7 @@ public class NoOpIndexClient
implements IndexClient
{
@Override
public List<IndexMetadata> readSplitIndex(String path, String... filterIndexTypes)
public List<IndexMetadata> 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");

10
presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java
View File

@ -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);

18
presto-parser/src/main/antlr4/io/prestosql/sql/parser/SqlBase.g4
View File

@ -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 : '<>' | '!=';

44
presto-parser/src/main/java/io/prestosql/sql/SqlFormatter.java
View File

@ -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)
{

52
presto-parser/src/main/java/io/prestosql/sql/parser/AstBuilder.java
View File

@ -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<Property> properties = ImmutableList.of();
if (context.properties() != null) {
properties = visit(context.properties().property(), Property.class);
}
List<Identifier> columnAliases = ImmutableList.of();
if (context.columnAliases() != null) {
columnAliases = visit(context.columnAliases().identifier(), Identifier.class);
}
Optional<String> 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)
{

15
presto-parser/src/main/java/io/prestosql/sql/tree/AstVisitor.java
View File

@ -582,6 +582,21 @@ public abstract class AstVisitor<R, C>
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);

150
presto-parser/src/main/java/io/prestosql/sql/tree/CreateIndex.java Normal file
View File

@ -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<Identifier> columnAliases;
private final String indexType;
private final boolean notExists;
private final List<Property> properties;
private final Optional<Expression> where;
public CreateIndex(QualifiedName indexName, QualifiedName tableName, List<Identifier> columnAliases,
String indexType, boolean notExists, List<Property> properties)
{
this(Optional.empty(), indexName, tableName, columnAliases, indexType, notExists, properties, Optional.empty());
}
public CreateIndex(NodeLocation location, QualifiedName indexName, QualifiedName tableName, List<Identifier> columnAliases,
String indexType, boolean notExists, List<Property> properties, Optional<Expression> where)
{
this(Optional.of(location), indexName, tableName, columnAliases, indexType, notExists, properties, where);
}
private CreateIndex(Optional<NodeLocation> location, QualifiedName indexName, QualifiedName tableName, List<Identifier> columnAliases,
String indexType, boolean notExists, List<Property> properties, Optional<Expression> 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<Identifier> getColumnAliases()
{
return columnAliases;
}
public String getIndexType()
{
return indexType;
}
public boolean isNotExists()
{
return notExists;
}
public List<Property> getProperties()
{
return properties;
}
public Optional<Expression> getExpression()
{
return where;
}
@Override
public <R, C> R accept(AstVisitor<R, C> visitor, C context)
{
return visitor.visitCreateIndex(this, context);
}
@Override
public List<Node> getChildren()
{
return ImmutableList.<Node>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();
}
}

97
presto-parser/src/main/java/io/prestosql/sql/tree/DropIndex.java Normal file
View File

@ -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<NodeLocation> 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, C> R accept(AstVisitor<R, C> visitor, C context)
{
return visitor.visitDropIndex(this, context);
}
@Override
public List<Node> 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();
}
}

82
presto-parser/src/main/java/io/prestosql/sql/tree/ShowIndex.java Normal file
View File

@ -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<QualifiedName> 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<NodeLocation> location, Optional<QualifiedName> indexName)
{
super(location);
this.indexName = indexName;
}
public QualifiedName getIndexName()
{
return indexName.isPresent() ? indexName.get() : null;
}
@Override
public <R, C> R accept(AstVisitor<R, C> visitor, C context)
{
return visitor.visitShowIndex(this, context);
}
@Override
public List<Node> 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();
}
}

2
presto-parser/src/test/java/io/prestosql/sql/parser/TestSqlParserErrorHandling.java
View File

@ -88,7 +88,7 @@ public class TestSqlParserErrorHandling
{"select foo(DISTINCT ,1)",
"line 1:21: mismatched input ','. Expecting: <expression>"},
{"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', <identifier>"},
{"CREATE TABLE foo ",

134
presto-spi/src/main/java/io/prestosql/spi/connector/CreateIndexMetadata.java Normal file
View File

@ -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<String> indexColumns;
private final String expression;
private final List<String> properties;
@JsonCreator
public CreateIndexMetadata(
@JsonProperty("indexName") String indexName,
@JsonProperty("tableName") String tableName,
@JsonProperty("indexType") String indexType,
@JsonProperty("indexColumns") List<String> indexColumns,
@JsonProperty("expression") String expression,
@JsonProperty("properties") List<String> 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<String> getIndexColumns()
{
return indexColumns;
}
@JsonProperty
public String getExpression()
{
return expression;
}
@JsonProperty
public List<String> 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);
}
}

3
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/DataSource.java
View File

@ -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);
}
}

13
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexClient.java
View File

@ -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<IndexMetadata> readSplitIndex(String path, String... filterIndexTypes) throws IOException;
public List<IndexMetadata> 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;
}

7
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexWriter.java
View File

@ -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;
}