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

!393 Store index properties in metadata and check connector type when creating index 

Merge pull request !393 from Han_Weng/store-index-properties
This commit is contained in:
i-robot 2020-11-24 00:50:12 +08:00 committed by Gitee
parent 4b9d2848e5 88dc9ae46c
commit 9fd979e3f7
19 changed files with 163 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -217,12 +217,17 @@ public class HeuristicIndexClient
public void addIndexRecord(CreateIndexMetadata createIndexMetadata)
throws IOException
{
List<String> properties = new LinkedList<>();
for (String propKey : createIndexMetadata.getProperties().stringPropertyNames()) {
properties.add(propKey + "=" + createIndexMetadata.getProperties().getProperty(propKey));
}
indexRecordManager.addIndexRecord(
createIndexMetadata.getIndexName(),
createIndexMetadata.getUser(),
createIndexMetadata.getTableName(),
createIndexMetadata.getIndexColumns().stream().map(Map.Entry::getKey).toArray(String[]::new),
createIndexMetadata.getIndexType(),
properties,
createIndexMetadata.getPartitions());
}

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

@ -15,7 +15,6 @@
package io.hetu.core.heuristicindex;
import com.google.common.collect.ImmutableSet;
import io.airlift.log.Logger;
import io.hetu.core.heuristicindex.filter.HeuristicIndexFilter;
import io.hetu.core.plugin.heuristicindex.index.bloom.BloomIndex;
@ -35,7 +34,6 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import static java.util.Objects.requireNonNull;
@ -49,7 +47,6 @@ public class HeuristicIndexFactory
implements IndexFactory
{
private static final Logger LOG = Logger.get(HeuristicIndexFactory.class);
private final Set<String> supportedConnector = ImmutableSet.of("hive");
public HeuristicIndexFactory()
{
@ -101,10 +98,4 @@ public class HeuristicIndexFactory
{
return new HeuristicIndexFilter(indices);
}
@Override
public Set<String> getSupportedConnector()
{
return supportedConnector;
}
}

11
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
View File

@ -87,6 +87,10 @@ public class IndexRecordManager
records.add(new IndexRecord(line));
}
}
catch (Exception e) {
throw new IllegalArgumentException(
"Error reading index record. Index record storage has been updated. Please delete old index directory and recreate the indices.");
}
cache = records;
cacheLastModifiedTime = modifiedTime;
}
@ -127,7 +131,7 @@ public class IndexRecordManager
* 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 synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
public synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
throws IOException
{
// Protect root directory
@ -143,7 +147,7 @@ public class IndexRecordManager
iterator.remove();
}
}
records.add(new IndexRecord(name, user, table, columns, indexType, partitions));
records.add(new IndexRecord(name, user, table, columns, indexType, indexProperties, partitions));
writeIndexRecords(records);
}
finally {
@ -179,7 +183,8 @@ public class IndexRecordManager
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", ImmutableList.of("Partitions")).toCsvRecord();
String head = new IndexRecord("Name", "User", "Table", new String[] {"Columns"}, "IndexType",
ImmutableList.of("Properties"), ImmutableList.of("Partitions")).toCsvRecord();
os.write(head.getBytes());
for (IndexRecord record : records) {
os.write(record.toCsvRecord().getBytes());

40
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java
View File

@ -51,14 +51,14 @@ public class TestIndexRecordManager
IndexRecordManager indexRecordManager1 = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
IndexRecordManager indexRecordManager2 = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
indexRecordManager1.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
indexRecordManager1.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
List<IndexRecord> original1 = indexRecordManager1.getIndexRecords();
List<IndexRecord> original2 = indexRecordManager2.getIndexRecords();
assertEquals(original2.size(), 1);
assertEquals(original1, original2);
List<IndexRecord> beforeadd1 = indexRecordManager1.getIndexRecords();
indexRecordManager2.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Arrays.asList("cp=1"));
indexRecordManager2.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList(), Arrays.asList("cp=1"));
List<IndexRecord> added2 = indexRecordManager2.getIndexRecords();
assertEquals(added2.size(), 2);
@ -89,7 +89,7 @@ public class TestIndexRecordManager
threads[i] = new Thread(() -> {
try {
new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath())
.addIndexRecord(names[finalI], "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
.addIndexRecord(names[finalI], "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException e) {
throw new RuntimeException(e);
@ -147,7 +147,7 @@ public class TestIndexRecordManager
int finalI = i;
threads[i] = new Thread(() -> {
try {
indexRecordManager.addIndexRecord(names[finalI], "u", "t", new String[] {"c"}, "minmax", Arrays.asList("cp=1"));
indexRecordManager.addIndexRecord(names[finalI], "u", "t", new String[] {"c"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException e) {
throw new RuntimeException(e);
@ -193,8 +193,8 @@ public class TestIndexRecordManager
try (TempFolder folder = new TempFolder()) {
folder.create();
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
indexRecordManager.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
indexRecordManager.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
indexRecordManager.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
indexRecordManager.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
assertNotNull(indexRecordManager.lookUpIndexRecord("1"));
assertEquals(indexRecordManager.getIndexRecords().size(), 2);
@ -221,21 +221,21 @@ public class TestIndexRecordManager
public void testAddAndLookUp()
throws IOException, IllegalAccessException
{
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn", "testColumn2"}, "minmax", Collections.emptyList());
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ImmutableList.of("12"));
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ImmutableList.of("12", "123"));
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Collections.emptyList());
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn", "testColumn2"}, "minmax", Collections.emptyList(), Collections.emptyList());
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("12"));
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("12", "123"));
}
@Test
public void testRecordEqualAndHash()
{
IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Collections.emptyList());
IndexRecord r2 = new IndexRecord("testName", "testUser", "testTable", new String[] {
"testColumn"}, "minmax", ImmutableList.of("note"));
IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList());
"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("note"));
IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList(), Collections.emptyList());
IndexRecord r4 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn",
"testColumn2"}, "minmax", Collections.emptyList());
"testColumn2"}, "minmax", Collections.emptyList(), Collections.emptyList());
assertEquals(r1, r1);
assertEquals(r1, r2);
assertNotEquals(r1, r3);
@ -253,28 +253,28 @@ public class TestIndexRecordManager
}
@Test(expectedExceptions = AssertionError.class)
public void testAddAndLookUpDifferentNotes()
public void testAddAndLookUpDifferentPartitions()
throws IOException, IllegalAccessException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
IndexRecord expected = new IndexRecord("testName", "testUser", "testTable", new String[] {
"testColumn"}, "minmax", ImmutableList.of(""));
indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of(""));
indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
IndexRecord actual = indexRecordManager.lookUpIndexRecord("testName");
assertIndexRecordFullyEqual(actual, expected);
}
}
private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, List<String> note)
private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
throws IOException, IllegalAccessException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
IndexRecord expected = new IndexRecord(name, user, table, columns, indexType, note);
indexRecordManager.addIndexRecord(name, user, table, columns, indexType, note);
IndexRecord expected = new IndexRecord(name, user, table, columns, indexType, indexProperties, partitions);
indexRecordManager.addIndexRecord(name, user, table, columns, indexType, indexProperties, partitions);
IndexRecord actual1 = indexRecordManager.lookUpIndexRecord(name);
assertNotNull(actual1);

6
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java
View File

@ -586,6 +586,12 @@ public class HiveMetadata
false));
}
@Override
public boolean isHeuristicIndexSupported()
{
return true;
}
@Override
public List<SchemaTableName> listTables(ConnectorSession session, Optional<String> optionalSchemaName)
{

22
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveTableHandle.java
View File

@ -31,6 +31,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static io.prestosql.plugin.hive.HiveMetadata.STORAGE_FORMAT;
import static java.util.Objects.requireNonNull;
@ -358,4 +359,25 @@ public class HiveTableHandle
{
return HiveStorageFormat.ORC.getOutputFormat().equals(tableParameters.get().get(STORAGE_FORMAT));
}
/**
* ORC is the only format supported to create heuristic index now
* We will add more formats in the future.
*/
@Override
public boolean isHeuristicIndexSupported()
{
return Stream.of(HiveStorageFormat.ORC)
.anyMatch(storageFormat -> storageFormat.getOutputFormat().equals(tableParameters.get().get(STORAGE_FORMAT)));
}
/**
* Create heuristic index... where predicate = xxx
* The predicate column only support partition columns
*/
@Override
public boolean isPartitionColumn(String column)
{
return partitionColumns.stream().map(HiveColumnHandle::getColumnName).collect(Collectors.toSet()).contains(column);
}
}

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

@ -57,7 +57,7 @@ public class TestIndexCache
private HiveColumnHandle partitionColumnHandle;
private HiveSplit testHiveSplit;
private List<HiveColumnHandle> testPartitions = Collections.emptyList();
private final long loadDelay = 0;
private final long loadDelay = 1000;
@BeforeClass
public void setupBeforeClass()

6
presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java
View File

@ -35,7 +35,6 @@ import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
public class HeuristicIndexerManager
{
@ -99,9 +98,4 @@ public class HeuristicIndexerManager
{
return factory.getIndexFilter(indices);
}
public Set<String> getSupportedCatalog()
{
return factory.getSupportedConnector();
}
}

8
presto-main/src/main/java/io/prestosql/metadata/Metadata.java
View File

@ -532,6 +532,14 @@ public interface Metadata
*/
boolean isExecutionPlanCacheSupported(Session session, TableHandle handle);
/**
* Hetu can only create index for supported connectors.
*
* @param session Presto session
* @param tableName Connector specific tableName
*/
boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName);
/**
* Hetu supports pushing sub-query with join down to the connector.
* This method decides if the sub-query can be pushed down to the connector based on the connector.

21
presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java
View File

@ -1062,6 +1062,27 @@ public final class MetadataManager
return metadata.isExecutionPlanCacheSupported(session.toConnectorSession(), table.getConnectorHandle());
}
/**
* Hetu can only create index for supported connectors.
*
* @param session Presto session
* @param tableName Connector specific tableName
*/
@Override
public boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName)
{
Optional<CatalogMetadata> catalog = getOptionalCatalogMetadata(session, tableName.getCatalogName());
if (catalog.isPresent()) {
CatalogMetadata catalogMetadata = catalog.get();
CatalogName catalogName = catalogMetadata.getConnectorId(session, tableName);
ConnectorMetadata metadata = catalogMetadata.getMetadataFor(catalogName);
return metadata.isHeuristicIndexSupported();
}
return false;
}
@Override
public Optional<LimitApplicationResult<TableHandle>> applyLimit(Session session, TableHandle table, long limit)
{

26
presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java
View File

@ -1031,17 +1031,35 @@ class StatementAnalyzer
CreateIndex createIndex = (CreateIndex) analysis.getOriginalStatement();
QualifiedObjectName tableFullName = MetadataUtil.createQualifiedObjectName(session, createIndex, createIndex.getTableName());
String tableName = tableFullName.toString();
// check catalog validate
if (!heuristicIndexerManager.getSupportedCatalog().contains(tableFullName.getCatalogName())) {
// check whether catalog support create index
if (!metadata.isHeuristicIndexSupported(session, tableFullName)) {
throw new SemanticException(NOT_SUPPORTED, createIndex,
"CREATE INDEX is not supported in '%s' connector",
"CREATE INDEX is not supported in catalog '%s'",
tableFullName.getCatalogName());
}
List<String> partitions = new ArrayList<>();
String partitionColumn = null;
if (createIndex.getExpression().isPresent()) {
partitions = HeuristicIndexUtils.extractPartitions(createIndex.getExpression().get());
// check partitions validate
// check partition name validate, create index …… where pt_d = xxx;
// pt_d must be partition column
List<String> partitionColumns = partitions.stream().map(k -> k.substring(0, k.indexOf("="))).collect(Collectors.toList());
if (partitionColumns.size() > 1) {
throw new IllegalArgumentException("Heuristic index only supports predicates on one column");
}
partitionColumn = partitionColumns.get(0);
}
Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableFullName);
if (tableHandle.isPresent() && !tableHandle.get().getConnectorHandle().isHeuristicIndexSupported()) {
throw new SemanticException(NOT_SUPPORTED, table, "Catalog supported, but table storage format is not supported by heuristic index");
}
if (tableHandle.isPresent() && partitionColumn != null
&& !tableHandle.get().getConnectorHandle().isPartitionColumn(partitionColumn)) {
throw new SemanticException(NOT_SUPPORTED, table, "Heuristic index creation is only supported for predicates on partition columns");
}
List<Map.Entry<String, Type>> indexColumns = new LinkedList<>();
for (Identifier i : createIndex.getColumnAliases()) {
indexColumns.add(new AbstractMap.SimpleEntry<>(i.toString(), BIGINT));

2
presto-main/src/main/java/io/prestosql/sql/planner/QueryPlanner.java
View File

@ -617,13 +617,13 @@ class QueryPlanner
Properties indexProperties = new Properties();
CreateIndexMetadata.Level indexCreationLevel = LEVEL_DEFAULT;
indexProperties.setProperty(LEVEL_PROP_KEY, String.valueOf(LEVEL_DEFAULT));
for (Property property : createIndex.getProperties()) {
String key = property.getName().toString().replaceAll("\"", "");
String val = property.getValue().toString().replaceAll("\"", "").toUpperCase(Locale.ENGLISH);
if (key.equals(LEVEL_PROP_KEY)) {
indexCreationLevel = CreateIndexMetadata.Level.valueOf(val);
continue;
}
indexProperties.setProperty(key, val);
}

2
presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java
View File

@ -668,7 +668,7 @@ final class ShowQueriesRewrite
new StringLiteral(String.join(",", v.columns)),
new StringLiteral(v.indexType),
new StringLiteral(partitionsStrToDisplay.toString()),
new StringLiteral(""),
new StringLiteral(String.join(",", v.properties)),
TRUE_LITERAL));
}

11
presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java
View File

@ -697,4 +697,15 @@ public abstract class AbstractMockMetadata
{
return true;
}
/**
* Hetu can only create index for supported connectors.
*
* @param session Presto session
* @param tableName Connector specific tableName
*/
public boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName)
{
return true;
}
}

8
presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java
View File

@ -922,4 +922,12 @@ public interface ConnectorMetadata
{
return false;
}
/**
* Hetu can only create index for supported connectors.
*/
default boolean isHeuristicIndexSupported()
{
return false;
}
}

12
presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorTableHandle.java
View File

@ -108,4 +108,16 @@ public interface ConnectorTableHandle
{
return false;
}
/* This method checks if heuristic index can be created with the table format */
default boolean isHeuristicIndexSupported()
{
return false;
}
/* This method checks if the predicate columns are partition columns */
default boolean isPartitionColumn(String column)
{
return false;
}
}

8
presto-spi/src/main/java/io/prestosql/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java
View File

@ -767,6 +767,14 @@ public class ClassLoaderSafeConnectorMetadata
return delegate.isExecutionPlanCacheSupported(session, handle);
}
/**
* Hetu can only create index for supported connectors.
*/
public boolean isHeuristicIndexSupported()
{
return delegate.isHeuristicIndexSupported();
}
@Override
public List<ConnectorVacuumTableInfo> getTablesForVacuum()
{

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

@ -21,7 +21,6 @@ import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
public interface IndexFactory
{
@ -47,6 +46,4 @@ public interface IndexFactory
public IndexClient getIndexClient(HetuFileSystemClient fs, Path root);
public IndexFilter getIndexFilter(Map<String, List<IndexMetadata>> indices);
public Set<String> getSupportedConnector();
}

14
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
View File

@ -15,9 +15,9 @@
package io.prestosql.spi.heuristicindex;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
public class IndexRecord
{
@ -27,32 +27,36 @@ public class IndexRecord
public final String table;
public final String[] columns;
public final String indexType;
public final List<String> properties;
public final List<String> partitions;
public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> properties, List<String> partitions)
{
this.name = name;
this.user = user == null ? "" : user;
this.table = table;
this.columns = columns;
this.indexType = indexType;
this.properties = properties;
this.partitions = partitions;
}
public IndexRecord(String csvRecord)
{
String[] records = csvRecord.split("\\t");
String[] records = csvRecord.split("\\|", Integer.MAX_VALUE);
this.name = records[0];
this.user = records[1];
this.table = records[2];
this.columns = records[3].split(COLUMN_DELIMITER);
this.indexType = records[4];
this.partitions = records.length > 5 ? Arrays.asList(records[5].split(",")) : Collections.emptyList();
this.properties = Arrays.stream(records[5].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
this.partitions = Arrays.stream(records[6].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
}
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, String.join(",", partitions));
return String.format("%s|%s|%s|%s|%s|%s|%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType,
String.join(",", properties), String.join(",", partitions));
}
@Override