diff --git a/presto-memory/pom.xml b/presto-memory/pom.xml index be4ff630f..3ca690526 100644 --- a/presto-memory/pom.xml +++ b/presto-memory/pom.xml @@ -107,6 +107,12 @@ + + io.hetu.core + hetu-common + test + + io.hetu.core presto-tests @@ -161,5 +167,17 @@ assertj-core test + + + io.hetu.core + hetu-metastore + test + + + + io.hetu.core + hetu-filesystem-client + test + diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/ColumnInfo.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/ColumnInfo.java index f0ae81421..a37e7e58c 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/ColumnInfo.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/ColumnInfo.java @@ -15,31 +15,28 @@ package io.prestosql.plugin.memory; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.connector.ColumnMetadata; import io.prestosql.spi.type.Type; +import io.prestosql.spi.type.TypeManager; import static java.util.Objects.requireNonNull; public class ColumnInfo { - private final ColumnHandle handle; + private final MemoryColumnHandle handle; private final String name; - private final Type type; @JsonCreator public ColumnInfo( - @JsonProperty("handle") ColumnHandle handle, - @JsonProperty("name") String name, - @JsonProperty("type") Type type) + @JsonProperty("handle") MemoryColumnHandle handle, + @JsonProperty("name") String name) { this.handle = requireNonNull(handle, "handle is null"); this.name = requireNonNull(name, "name is null"); - this.type = requireNonNull(type, "type is null"); } @JsonProperty - public ColumnHandle getHandle() + public MemoryColumnHandle getHandle() { return handle; } @@ -50,25 +47,24 @@ public class ColumnInfo return name; } - @JsonProperty - public Type getType() + public Type getType(TypeManager typeManager) { - return type; + return handle.getType(typeManager); } - public ColumnMetadata getMetadata() + public ColumnMetadata getMetadata(TypeManager typeManager) { - return new ColumnMetadata(name, type); + return new ColumnMetadata(name, getType(typeManager)); } public int getIndex() { - return ((MemoryColumnHandle) handle).getColumnIndex(); + return handle.getColumnIndex(); } @Override public String toString() { - return name + "::" + type; + return name + "::" + handle.getTypeSignature(); } } diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryColumnHandle.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryColumnHandle.java index e54db1759..74406aa6b 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryColumnHandle.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryColumnHandle.java @@ -17,6 +17,8 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.type.Type; +import io.prestosql.spi.type.TypeManager; +import io.prestosql.spi.type.TypeSignature; import java.util.Objects; @@ -24,14 +26,15 @@ public final class MemoryColumnHandle implements ColumnHandle { private final int columnIndex; - private final Type type; + private final TypeSignature typeSignature; + private Type typeCache; @JsonCreator public MemoryColumnHandle(@JsonProperty("columnIndex") int columnIndex, - @JsonProperty("type") Type type) + @JsonProperty("typeSignature") TypeSignature typeSignature) { this.columnIndex = columnIndex; - this.type = type; + this.typeSignature = typeSignature; } @JsonProperty @@ -41,9 +44,17 @@ public final class MemoryColumnHandle } @JsonProperty - public Type getType() + public TypeSignature getTypeSignature() { - return type; + return typeSignature; + } + + public Type getType(TypeManager typeManager) + { + if (typeCache == null) { + typeCache = typeManager.getType(getTypeSignature()); + } + return typeCache; } @Override diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryConnectorFactory.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryConnectorFactory.java index 7f1c7c86a..6ac65d5bc 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryConnectorFactory.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryConnectorFactory.java @@ -23,6 +23,7 @@ import io.prestosql.spi.connector.ConnectorFactory; import io.prestosql.spi.connector.ConnectorHandleResolver; import io.prestosql.spi.function.FunctionMetadataManager; import io.prestosql.spi.function.StandardFunctionResolution; +import io.prestosql.spi.metastore.HetuMetastore; import io.prestosql.spi.relation.DeterminismEvaluator; import io.prestosql.spi.relation.RowExpressionService; import io.prestosql.spi.type.TypeManager; @@ -61,6 +62,7 @@ public class MemoryConnectorFactory binder.bind(TypeManager.class).toInstance(context.getTypeManager()); binder.bind(RowExpressionService.class).toInstance(context.getRowExpressionService()); binder.bind(DeterminismEvaluator.class).toInstance(context.getRowExpressionService().getDeterminismEvaluator()); + binder.bind(HetuMetastore.class).toInstance(context.getHetuMetastore()); }, new JsonModule(), new MemoryModule(context.getTypeManager(), context.getNodeManager())); diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java index 42dc0f7ac..50eeeff4a 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java @@ -14,9 +14,7 @@ package io.prestosql.plugin.memory; import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableMap; -import com.google.common.collect.ImmutableSet; -import com.google.common.collect.Maps; +import io.airlift.json.JsonCodec; import io.airlift.slice.Slice; import io.prestosql.spi.HostAddress; import io.prestosql.spi.Node; @@ -39,13 +37,20 @@ import io.prestosql.spi.connector.ConstraintApplicationResult; import io.prestosql.spi.connector.SchemaNotFoundException; import io.prestosql.spi.connector.SchemaTableName; import io.prestosql.spi.connector.SchemaTablePrefix; +import io.prestosql.spi.connector.TableNotFoundException; import io.prestosql.spi.connector.ViewNotFoundException; +import io.prestosql.spi.metastore.HetuMetastore; +import io.prestosql.spi.metastore.model.CatalogEntity; +import io.prestosql.spi.metastore.model.DatabaseEntity; +import io.prestosql.spi.metastore.model.TableEntity; +import io.prestosql.spi.metastore.model.TableEntityType; import io.prestosql.spi.statistics.ComputedStatistics; +import io.prestosql.spi.type.TypeManager; import javax.annotation.concurrent.ThreadSafe; import javax.inject.Inject; -import java.util.ArrayList; +import java.util.Base64; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -56,103 +61,127 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; +import java.util.stream.Stream; import static com.google.common.base.Preconditions.checkState; -import static com.google.common.base.Verify.verify; -import static com.google.common.collect.ImmutableList.toImmutableList; +import static io.airlift.json.JsonCodec.jsonCodec; import static io.prestosql.spi.StandardErrorCode.ALREADY_EXISTS; import static io.prestosql.spi.StandardErrorCode.INVALID_TABLE_PROPERTY; -import static io.prestosql.spi.StandardErrorCode.NOT_FOUND; import static io.prestosql.spi.StandardErrorCode.SCHEMA_NOT_EMPTY; import static java.lang.String.format; import static java.util.Objects.requireNonNull; -import static java.util.stream.Collectors.toList; import static java.util.stream.Collectors.toMap; @ThreadSafe public class MemoryMetadata implements ConnectorMetadata { - public static final String SCHEMA_NAME = "default"; + public static final String MEM_KEY = "memory"; + public static final String DEFAULT_SCHEMA = "default"; + public static final String ID_KEY = "id"; + public static final String NEXT_ID_KEY = "_NEXT_ID_"; + private static final JsonCodec VIEW_CODEC = jsonCodec(ConnectorViewDefinition.class); private final NodeManager nodeManager; - private final List schemas = new ArrayList<>(); - private final AtomicLong nextTableId = new AtomicLong(); - private final Map tableIds = new ConcurrentHashMap<>(); + private final TypeManager typeManager; + private final AtomicLong nextTableId; private final Map tables = new ConcurrentHashMap<>(); private final Map views = new ConcurrentHashMap<>(); + private final HetuMetastore metastore; @Inject - public MemoryMetadata(NodeManager nodeManager) + public MemoryMetadata(TypeManager typeManager, NodeManager nodeManager, HetuMetastore metastore) { + this.typeManager = requireNonNull(typeManager, "typeManager is null"); this.nodeManager = requireNonNull(nodeManager, "nodeManager is null"); - this.schemas.add(SCHEMA_NAME); + this.metastore = metastore; + Optional oldCatalog = metastore.getCatalog(MEM_KEY); + if (!oldCatalog.isPresent()) { + CatalogEntity newCatalog = CatalogEntity.builder() + .setCatalogName(MEM_KEY) + .setComment(Optional.of("Hetu memory connector")) + .build(); + metastore.createCatalogIfNotExist(newCatalog); + this.nextTableId = new AtomicLong(); + } + else { + this.nextTableId = new AtomicLong(Long.parseLong(oldCatalog.get().getParameters().getOrDefault(NEXT_ID_KEY, "0"))); + } + DatabaseEntity.Builder databaseBuilder = DatabaseEntity.builder() + .setCatalogName(MEM_KEY) + .setDatabaseName(DEFAULT_SCHEMA); + metastore.createDatabaseIfNotExist(databaseBuilder.build()); } @Override public synchronized List listSchemaNames(ConnectorSession session) { - return ImmutableList.copyOf(schemas); + ImmutableList.Builder schemaNames = ImmutableList.builder(); + metastore.getAllDatabases(MEM_KEY).forEach(databaseEntity -> schemaNames.add(databaseEntity.getName())); + return schemaNames.build(); } @Override public synchronized void createSchema(ConnectorSession session, String schemaName, Map properties) { - if (schemas.contains(schemaName)) { - throw new PrestoException(ALREADY_EXISTS, format("Schema [%s] already exists", schemaName)); - } - schemas.add(schemaName); + checkSchemaNotExists(schemaName); + + DatabaseEntity.Builder databaseBuilder = DatabaseEntity.builder() + .setCatalogName(MEM_KEY) + .setDatabaseName(schemaName) + .setCreateTime(session.getStartTime()) + .setOwner(session.getUser()); + metastore.createDatabaseIfNotExist(databaseBuilder.build()); } @Override public synchronized void dropSchema(ConnectorSession session, String schemaName) { - if (!schemas.contains(schemaName)) { - throw new PrestoException(NOT_FOUND, format("Schema [%s] does not exist", schemaName)); - } + checkSchemaExists(schemaName); - boolean tablesExist = tables.values().stream() - .anyMatch(table -> table.getSchemaName().equals(schemaName)); - - if (tablesExist) { + if (!metastore.getAllTables(MEM_KEY, schemaName).isEmpty()) { throw new PrestoException(SCHEMA_NOT_EMPTY, "Schema not empty: " + schemaName); } - verify(schemas.remove(schemaName)); + metastore.dropDatabase(MEM_KEY, schemaName); } @Override public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) { - Long id = tableIds.get(schemaTableName); - if (id == null) { + Optional tableEntity = metastore.getTable(MEM_KEY, schemaTableName.getSchemaName(), schemaTableName.getTableName()); + if (!tableEntity.isPresent()) { return null; } - return new MemoryTableHandle(id); + String idStr = tableEntity.get().getParameters().get(ID_KEY); + if (idStr == null) { + return null; + } + + return new MemoryTableHandle(Long.parseLong(idStr)); } @Override public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle) { MemoryTableHandle handle = (MemoryTableHandle) tableHandle; - return tables.get(handle.getId()).getMetadata(); + return getTableInfo((handle).getId()).getMetadata(typeManager); } @Override public synchronized List listTables(ConnectorSession session, Optional schemaName) { - return tables.values().stream() - .filter(table -> schemaName.map(table.getSchemaName()::equals).orElse(true)) - .map(TableInfo::getSchemaTableName) - .collect(toList()); + return getTableStream(schemaName, false) + .collect(Collectors.toList()); } @Override public Map getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle) { MemoryTableHandle handle = (MemoryTableHandle) tableHandle; - return tables.get(handle.getId()) + return getTableInfo(handle.getId()) .getColumns().stream() .collect(toMap(ColumnInfo::getName, ColumnInfo::getHandle)); } @@ -161,49 +190,61 @@ public class MemoryMetadata public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle) { MemoryTableHandle handle = (MemoryTableHandle) tableHandle; - return tables.get(handle.getId()) + return getTableInfo(handle.getId()) .getColumn(columnHandle) - .getMetadata(); + .getMetadata(typeManager); } @Override public Map> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix) { - return tables.values().stream() + return getMemoryCatalogEntity().getParameters().values().stream() + .map(TableInfo::deserialize) .filter(table -> prefix.matches(table.getSchemaTableName())) - .collect(toMap(TableInfo::getSchemaTableName, handle -> handle.getMetadata().getColumns())); + .collect(toMap(TableInfo::getSchemaTableName, handle -> handle.getMetadata(typeManager).getColumns())); } @Override public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle) { MemoryTableHandle handle = (MemoryTableHandle) tableHandle; - TableInfo info = tables.remove(handle.getId()); - if (info != null) { - tableIds.remove(info.getSchemaTableName()); - } + TableInfo info = getTableInfo(handle.getId()); + metastore.dropTable(MEM_KEY, info.getSchemaName(), info.getTableName()); + updateTableInfo(handle.getId(), null); } + // CONTINUE HERE + @Override public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName) { checkSchemaExists(newTableName.getSchemaName()); - checkTableNotExists(newTableName); + checkTableNotExists(newTableName, false); MemoryTableHandle handle = (MemoryTableHandle) tableHandle; long tableId = handle.getId(); - TableInfo oldInfo = tables.get(tableId); - tables.put(tableId, new TableInfo(tableId, newTableName.getSchemaName(), newTableName.getTableName(), oldInfo.getColumns(), oldInfo.getDataFragments())); + TableInfo oldInfo = getTableInfo(tableId); + updateTableInfo(tableId, new TableInfo( + tableId, + newTableName.getSchemaName(), + newTableName.getTableName(), + oldInfo.getColumns(), + oldInfo.getDataFragments())); - tableIds.remove(oldInfo.getSchemaTableName()); - tableIds.put(newTableName, tableId); + metastore.alterTable(MEM_KEY, oldInfo.getSchemaName(), oldInfo.getTableName(), + TableEntity.builder() + .setCatalogName(MEM_KEY) + .setDatabaseName(newTableName.getSchemaName()) + .setTableName(newTableName.getTableName()) + .setTableType(TableEntityType.TABLE.toString()) + .setParameter(ID_KEY, String.valueOf(tableId)) + .build()); } @Override public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting) { - tableMetadata.getProperties(); ConnectorOutputTableHandle outputTableHandle = beginCreateTable(session, tableMetadata, Optional.empty()); finishCreateTable(session, outputTableHandle, ImmutableList.of(), ImmutableList.of()); } @@ -212,7 +253,7 @@ public class MemoryMetadata public MemoryOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional layout) { checkSchemaExists(tableMetadata.getTable().getSchemaName()); - checkTableNotExists(tableMetadata.getTable()); + checkTableNotExists(tableMetadata.getTable(), false); List sortedBy = MemoryTableProperties.getSortedBy(tableMetadata.getProperties()); if (sortedBy == null) { @@ -250,7 +291,7 @@ public class MemoryMetadata Set columnNames = new HashSet<>(); for (int i = 0; i < tableMetadata.getColumns().size(); i++) { ColumnMetadata column = tableMetadata.getColumns().get(i); - columns.add(new ColumnInfo(new MemoryColumnHandle(i, column.getType()), column.getName(), column.getType())); + columns.add(new ColumnInfo(new MemoryColumnHandle(i, column.getType().getTypeSignature()), column.getName())); columnNames.add(column.getName()); } @@ -265,38 +306,60 @@ public class MemoryMetadata } long nextId = nextTableId.getAndIncrement(); + metastore.alterCatalogParameter(MEM_KEY, NEXT_ID_KEY, String.valueOf(nextTableId.get())); Set nodes = nodeManager.getRequiredWorkerNodes(); checkState(!nodes.isEmpty(), "No Memory nodes available"); long tableId = nextId; List columnInfos = columns.build(); - tableIds.put(tableMetadata.getTable(), tableId); - tables.put(tableId, new TableInfo( + metastore.createTable(TableEntity.builder() + .setCatalogName(MEM_KEY) + .setDatabaseName(tableMetadata.getTable().getSchemaName()) + .setTableName(tableMetadata.getTable().getTableName()) + .setTableType(TableEntityType.TABLE.toString()) + .setParameter(ID_KEY, String.valueOf(tableId)) + .build()); + updateTableInfo(tableId, new TableInfo( tableId, tableMetadata.getTable().getSchemaName(), tableMetadata.getTable().getTableName(), columnInfos, new HashMap<>())); - return new MemoryOutputTableHandle(tableId, ImmutableSet.copyOf(tableIds.values()), columnInfos, sortedBy, indexColumns); + return new MemoryOutputTableHandle(tableId, getTableIdSet(), columnInfos, sortedBy, indexColumns); } - private void checkSchemaExists(String schemaName) + private void checkSchemaNotExists(String schemaName) { - if (!schemas.contains(schemaName)) { + if (metastore.getDatabase(MEM_KEY, schemaName).isPresent()) { + throw new PrestoException(ALREADY_EXISTS, format("Schema already exists", schemaName)); + } + } + + private DatabaseEntity checkSchemaExists(String schemaName) + { + Optional databaseEntity = metastore.getDatabase(MEM_KEY, schemaName); + if (!databaseEntity.isPresent()) { throw new SchemaNotFoundException(schemaName); } + return databaseEntity.get(); } - private void checkTableNotExists(SchemaTableName tableName) + private void checkTableNotExists(SchemaTableName tableName, boolean isView) { - if (tableIds.containsKey(tableName)) { - throw new PrestoException(ALREADY_EXISTS, format("Table [%s] already exists", tableName.toString())); + if (metastore.getTable(MEM_KEY, tableName.getSchemaName(), tableName.getTableName()).isPresent()) { + throw new PrestoException(ALREADY_EXISTS, format("%s [%s] already exists", isView ? "View" : "Table", tableName)); } - if (views.containsKey(tableName)) { - throw new PrestoException(ALREADY_EXISTS, format("View [%s] already exists", tableName.toString())); + } + + private TableEntity checkTableExists(SchemaTableName tableName, boolean isView) + { + Optional tableEntity = metastore.getTable(MEM_KEY, tableName.getSchemaName(), tableName.getTableName()); + if (!tableEntity.isPresent()) { + throw isView ? new ViewNotFoundException(tableName) : new TableNotFoundException(tableName); } + return tableEntity.get(); } @Override @@ -313,7 +376,7 @@ public class MemoryMetadata public MemoryInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle) { MemoryTableHandle memoryTableHandle = (MemoryTableHandle) tableHandle; - return new MemoryInsertTableHandle(memoryTableHandle.getId(), ImmutableSet.copyOf(tableIds.values())); + return new MemoryInsertTableHandle(memoryTableHandle.getId(), getTableIdSet()); } @Override @@ -330,50 +393,47 @@ public class MemoryMetadata public void createView(ConnectorSession session, SchemaTableName viewName, ConnectorViewDefinition definition, boolean replace) { checkSchemaExists(viewName.getSchemaName()); - if (tableIds.containsKey(viewName)) { - throw new PrestoException(ALREADY_EXISTS, "Table already exists: " + viewName); - } - if (replace) { - views.put(viewName, definition); - } - else if (views.putIfAbsent(viewName, definition) != null) { - throw new PrestoException(ALREADY_EXISTS, "View already exists: " + viewName); + if (!replace) { + checkTableNotExists(viewName, true); } + updateViewDef(viewName, definition); } @Override public void dropView(ConnectorSession session, SchemaTableName viewName) { - if (views.remove(viewName) == null) { - throw new ViewNotFoundException(viewName); - } + checkTableExists(viewName, true); + updateViewDef(viewName, null); } @Override public List listViews(ConnectorSession session, Optional schemaName) { - return views.keySet().stream() - .filter(viewName -> schemaName.map(viewName.getSchemaName()::equals).orElse(true)) - .collect(toImmutableList()); + return getTableStream(schemaName, true) + .collect(Collectors.toList()); } @Override public Map getViews(ConnectorSession session, Optional schemaName) { SchemaTablePrefix prefix = schemaName.map(SchemaTablePrefix::new).orElseGet(SchemaTablePrefix::new); - return ImmutableMap.copyOf(Maps.filterKeys(views, prefix::matches)); + return getTableStream(schemaName, true) + .filter(prefix::matches) + .collect(Collectors.toMap( + name -> name, + this::getViewDef)); } @Override public Optional getView(ConnectorSession session, SchemaTableName viewName) { - return Optional.ofNullable(views.get(viewName)); + return Optional.ofNullable(getViewDef(viewName)); } private void updateRowsOnHosts(long tableId, Collection fragments) { - TableInfo info = tables.get(tableId); + TableInfo info = getTableInfo(tableId); checkState( info != null, "Uninitialized tableId [%s.%s]", @@ -386,7 +446,7 @@ public class MemoryMetadata dataFragments.merge(memoryDataFragment.getHostAddress(), memoryDataFragment, MemoryDataFragment::merge); } - tables.put(tableId, new TableInfo(tableId, info.getSchemaName(), info.getTableName(), info.getColumns(), dataFragments)); + updateTableInfo(tableId, new TableInfo(tableId, info.getSchemaName(), info.getTableName(), info.getColumns(), dataFragments)); } @Override @@ -403,7 +463,7 @@ public class MemoryMetadata public List getDataFragments(long tableId) { - return ImmutableList.copyOf(tables.get(tableId).getDataFragments().values()); + return ImmutableList.copyOf(getTableInfo(tableId).getDataFragments().values()); } // TODO: disabled for now @@ -451,4 +511,108 @@ public class MemoryMetadata return Optional.of(new ConstraintApplicationResult<>(newMemoryTableHandle, constraint.getSummary())); } + + /** + * Get all tables in the given schema (if present) or all tables in memory catalog. + */ + private Stream getTableStream(Optional schemaName, boolean isView) + { + Stream tables = schemaName.isPresent() ? + metastore.getAllTables(MEM_KEY, schemaName.get()).stream() : + metastore.getAllDatabases(MEM_KEY).stream() + .flatMap(databaseEntity -> metastore.getAllTables(MEM_KEY, databaseEntity.getName()).stream()); + return tables + .filter(tableEntity -> isView(tableEntity) == isView) + .map(tableEntity -> new SchemaTableName(tableEntity.getDatabaseName(), tableEntity.getName())); + } + + private TableInfo getTableInfo(Long tableId) + { + // cache tableInfo in the map to avoid deserializing it on every visit + TableInfo tableInfo = tables.get(tableId); + if (tableInfo == null) { + tableInfo = TableInfo.deserialize(getMemoryCatalogEntity().getParameters().get(tableId.toString())); + tables.put(tableId, tableInfo); + } + return tableInfo; + } + + /** + * Update catalog parameter map. remove entry if {@code null} tableInfo passed in. + */ + private void updateTableInfo(Long tableId, TableInfo newTableInfo) + { + if (newTableInfo == null) { + metastore.alterCatalogParameter(MEM_KEY, String.valueOf(tableId), null); + } + else { + metastore.alterCatalogParameter(MEM_KEY, String.valueOf(tableId), newTableInfo.serialize()); + } + tables.remove(tableId); + } + + private ConnectorViewDefinition getViewDef(SchemaTableName viewName) + { + // cache view def in the map to avoid deserializing it on every visit + ConnectorViewDefinition view = views.get(viewName); + if (view == null) { + try { + TableEntity tableEntity = checkTableExists(viewName, true); + if (isView(tableEntity)) { + view = VIEW_CODEC.fromJson(Base64.getDecoder().decode(tableEntity.getViewOriginalText())); + views.put(viewName, view); + } + } + catch (Exception e) { + views.remove(viewName); + } + } + return view; + } + + private void updateViewDef(SchemaTableName viewName, ConnectorViewDefinition newViewDef) + { + views.remove(viewName); + if (newViewDef == null) { + metastore.dropTable(MEM_KEY, viewName.getSchemaName(), viewName.getTableName()); + } + else { + String encoded = Base64.getEncoder().encodeToString(VIEW_CODEC.toJsonBytes(newViewDef)); + Optional tableEntity = metastore.getTable(MEM_KEY, viewName.getSchemaName(), viewName.getTableName()); + if (!tableEntity.isPresent()) { + metastore.createTable( + TableEntity.builder() + .setCatalogName(MEM_KEY) + .setDatabaseName(viewName.getSchemaName()) + .setTableName(viewName.getTableName()) + .setViewOriginalText(Optional.ofNullable(encoded)) + .setTableType(TableEntityType.TABLE.toString()) + .build()); + } + else { + TableEntity newTableEntity = tableEntity.get(); + newTableEntity.setViewOriginalText(encoded); + metastore.alterTable(MEM_KEY, viewName.getSchemaName(), viewName.getTableName(), newTableEntity); + } + } + } + + private CatalogEntity getMemoryCatalogEntity() + { + Optional catalogEntity = metastore.getCatalog(MEM_KEY); + if (!catalogEntity.isPresent()) { + throw new IllegalStateException("Metastore catalog " + MEM_KEY + " does not exist"); + } + return catalogEntity.get(); + } + + private boolean isView(TableEntity tableEntity) + { + return tableEntity.getViewOriginalText() != null; + } + + private Set getTableIdSet() + { + return getMemoryCatalogEntity().getParameters().keySet().stream().filter(e -> !NEXT_ID_KEY.equals(e)).map(Long::valueOf).collect(Collectors.toSet()); + } } diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java index b1812219d..43fe1f0fc 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java @@ -42,12 +42,14 @@ import static java.util.stream.Collectors.toList; public final class MemoryPageSourceProvider implements ConnectorPageSourceProvider { + private final TypeManager typeManager; private final MemoryPagesStore pagesStore; @Inject public MemoryPageSourceProvider(MemoryPagesStore pagesStore, TypeManager typeManager, MemoryMetadata memoryMetadata) { this.pagesStore = requireNonNull(pagesStore, "pagesStore is null"); + this.typeManager = requireNonNull(typeManager, "typeManager is null"); } @Override diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java index 5288357a2..e84750778 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java @@ -13,28 +13,42 @@ */ package io.prestosql.plugin.memory; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import io.airlift.json.JsonCodec; import io.prestosql.spi.HostAddress; import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.connector.ConnectorTableMetadata; import io.prestosql.spi.connector.SchemaTableName; +import io.prestosql.spi.type.TypeManager; +import java.util.Base64; import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import static io.airlift.json.JsonCodec.jsonCodec; import static java.util.Objects.requireNonNull; public class TableInfo { + private static final JsonCodec TABLE_INFO_JSON_CODEC = jsonCodec(TableInfo.class); + private final long id; private final String schemaName; private final String tableName; private final List columns; private final Map dataFragments; - public TableInfo(long id, String schemaName, String tableName, List columns, Map dataFragments) + @JsonCreator + public TableInfo( + @JsonProperty("id") long id, + @JsonProperty("schemaName") String schemaName, + @JsonProperty("tableName") String tableName, + @JsonProperty("columns") List columns, + @JsonProperty("dataFragments") Map dataFragments) { this.id = requireNonNull(id, "handle is null"); this.schemaName = requireNonNull(schemaName, "schemaName is null"); @@ -43,16 +57,19 @@ public class TableInfo this.dataFragments = ImmutableMap.copyOf(dataFragments); } + @JsonProperty public long getId() { return id; } + @JsonProperty public String getSchemaName() { return schemaName; } + @JsonProperty public String getTableName() { return tableName; @@ -63,15 +80,16 @@ public class TableInfo return new SchemaTableName(schemaName, tableName); } - public ConnectorTableMetadata getMetadata() + public ConnectorTableMetadata getMetadata(TypeManager typeManager) { return new ConnectorTableMetadata( new SchemaTableName(schemaName, tableName), columns.stream() - .map(ColumnInfo::getMetadata) + .map(columnInfo -> columnInfo.getMetadata(typeManager)) .collect(Collectors.toList())); } + @JsonProperty public List getColumns() { return columns; @@ -85,8 +103,19 @@ public class TableInfo .get(); } + @JsonProperty public Map getDataFragments() { return dataFragments; } + + public String serialize() + { + return Base64.getEncoder().encodeToString(TABLE_INFO_JSON_CODEC.toJsonBytes(this)); + } + + public static TableInfo deserialize(String serializedTableInfo) + { + return TABLE_INFO_JSON_CODEC.fromJson(Base64.getDecoder().decode(serializedTableInfo)); + } } diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java index eada5f997..e2fa624fd 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java @@ -30,6 +30,7 @@ import io.prestosql.spi.predicate.Range; import io.prestosql.spi.predicate.SortedRangeSet; import io.prestosql.spi.predicate.TupleDomain; import io.prestosql.spi.type.Type; +import io.prestosql.spi.type.TypeManager; import io.prestosql.spi.type.TypeUtils; import io.prestosql.spi.util.BloomFilter; @@ -75,7 +76,7 @@ public class LogicalPart private final Map indexChannelFilters = new HashMap<>(); private final Map> columnMinMax = new HashMap<>(); - public LogicalPart(List columns, List sortedBy, List indexColumns, PageSorter pageSorter, long maxLogicalPartBytes) + public LogicalPart(List columns, List sortedBy, List indexColumns, PageSorter pageSorter, long maxLogicalPartBytes, TypeManager typeManager) { requireNonNull(columns, "columns is null"); requireNonNull(sortedBy, "sortedBy is null"); @@ -85,7 +86,7 @@ public class LogicalPart types = new ArrayList<>(); for (ColumnInfo column : columns) { - types.add(column.getType()); + types.add(column.getType(typeManager)); } sortChannels = new ArrayList<>(); diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryPagesStore.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryPagesStore.java index b0c51cf8e..5790c7e33 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryPagesStore.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryPagesStore.java @@ -23,6 +23,7 @@ import io.prestosql.spi.PrestoException; import io.prestosql.spi.block.Block; import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.predicate.TupleDomain; +import io.prestosql.spi.type.TypeManager; import javax.annotation.concurrent.GuardedBy; import javax.annotation.concurrent.ThreadSafe; @@ -49,25 +50,27 @@ public class MemoryPagesStore private final int splitsPerNode; private final PageSorter pageSorter; private final MemoryConfig config; + private final TypeManager typeManager; @GuardedBy("this") private long currentBytes; private final Map tables = new HashMap<>(); @Inject - public MemoryPagesStore(MemoryConfig config, PageSorter pageSorter) + public MemoryPagesStore(MemoryConfig config, PageSorter pageSorter, TypeManager typeManager) { requireNonNull(pageSorter, "config is null"); this.pageSorter = requireNonNull(pageSorter, "pageSorter is null"); this.config = requireNonNull(config, "config is null"); this.splitsPerNode = config.getSplitsPerNode(); this.maxBytes = config.getMaxDataPerNode().toBytes(); + this.typeManager = requireNonNull(typeManager, "typeManager is null"); } public synchronized void initialize(long tableId, List columns, List sortedBy, List indexColumns) { if (!tables.containsKey(tableId)) { - tables.put(tableId, new TableData(tableId, columns, sortedBy, indexColumns, pageSorter, config)); + tables.put(tableId, new TableData(tableId, columns, sortedBy, indexColumns, pageSorter, config, typeManager)); } } diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/TableData.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/TableData.java index eb29f05dd..8a480f401 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/TableData.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/TableData.java @@ -23,6 +23,7 @@ import io.prestosql.spi.Page; import io.prestosql.spi.PageSorter; import io.prestosql.spi.connector.ColumnHandle; import io.prestosql.spi.predicate.TupleDomain; +import io.prestosql.spi.type.TypeManager; import java.util.ArrayList; import java.util.Arrays; @@ -55,10 +56,11 @@ public class TableData private final PageSorter pageSorter; private final long maxLogicalPartBytes; private final MemoryConfig config; + private final TypeManager typeManager; private long lastModified = System.currentTimeMillis(); public TableData(long id, List columns, List sortedBy, - List indexColumns, PageSorter pageSorter, MemoryConfig config) + List indexColumns, PageSorter pageSorter, MemoryConfig config, TypeManager typeManager) { this.id = requireNonNull(id, "id is null"); this.config = requireNonNull(config, "config is null"); @@ -68,6 +70,7 @@ public class TableData this.sortedBy = requireNonNull(sortedBy, "sortedBy is null"); this.indexColumns = requireNonNull(indexColumns, "indexColumns is null"); this.pageSorter = requireNonNull(pageSorter, "pageSorter is null"); + this.typeManager = requireNonNull(typeManager, "typeManager is null"); this.splits = new ArrayList[totalSplits]; for (int i = 0; i < totalSplits; i++) { @@ -111,7 +114,7 @@ public class TableData { List splitParts = splits[nextSplit.getAndIncrement() % totalSplits]; if (splitParts.isEmpty() || !splitParts.get(splitParts.size() - 1).canAdd()) { - splitParts.add(new LogicalPart(columns, sortedBy, indexColumns, pageSorter, maxLogicalPartBytes)); + splitParts.add(new LogicalPart(columns, sortedBy, indexColumns, pageSorter, maxLogicalPartBytes, typeManager)); } LogicalPart currentSplitPart = splitParts.get(splitParts.size() - 1); diff --git a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryMetadata.java b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryMetadata.java index d69759f2e..b91a3f710 100644 --- a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryMetadata.java +++ b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryMetadata.java @@ -16,6 +16,11 @@ package io.prestosql.plugin.memory; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; +import io.hetu.core.common.filesystem.TempFolder; +import io.hetu.core.filesystem.HetuLocalFileSystemClient; +import io.hetu.core.filesystem.LocalConfig; +import io.hetu.core.metastore.hetufilesystem.HetuFsMetastore; +import io.hetu.core.metastore.hetufilesystem.HetuFsMetastoreConfig; import io.prestosql.spi.PrestoException; import io.prestosql.spi.connector.ConnectorOutputTableHandle; import io.prestosql.spi.connector.ConnectorTableHandle; @@ -24,10 +29,13 @@ import io.prestosql.spi.connector.ConnectorViewDefinition; import io.prestosql.spi.connector.ConnectorViewDefinition.ViewColumn; import io.prestosql.spi.connector.SchemaNotFoundException; import io.prestosql.spi.connector.SchemaTableName; +import io.prestosql.spi.type.testing.TestingTypeManager; import io.prestosql.testing.TestingNodeManager; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import java.io.IOException; +import java.nio.file.Paths; import java.util.List; import java.util.Map; import java.util.Optional; @@ -51,8 +59,13 @@ public class TestMemoryMetadata @BeforeMethod public void setUp() + throws IOException { - metadata = new MemoryMetadata(new TestingNodeManager()); + TempFolder tmp = new TempFolder().create(); + Runtime.getRuntime().addShutdownHook(new Thread(tmp::close)); + metadata = new MemoryMetadata(new TestingTypeManager(), new TestingNodeManager(), + new HetuFsMetastore(new HetuFsMetastoreConfig().setHetuFileSystemMetastorePath(tmp.getRoot().getCanonicalPath()), + new HetuLocalFileSystemClient(new LocalConfig(null), Paths.get(tmp.getRoot().getCanonicalPath())))); } @Test @@ -70,8 +83,8 @@ public class TestMemoryMetadata metadata.finishCreateTable(SESSION, table, ImmutableList.of(), ImmutableList.of()); List tables = metadata.listTables(SESSION, Optional.empty()); - assertTrue(tables.size() == 1, "Expected only one table"); - assertTrue(tables.get(0).getTableName().equals("temp_table"), "Expected table with name 'temp_table'"); + assertEquals(tables.size(), 1, "Expected only one table"); + assertEquals(tables.get(0).getTableName(), "temp_table", "Expected table with name 'temp_table'"); } @Test @@ -169,7 +182,7 @@ public class TestMemoryMetadata assertEquals(metadata.listTables(SESSION, Optional.of("default")), ImmutableList.of()); } - @Test(expectedExceptions = PrestoException.class, expectedExceptionsMessageRegExp = "View already exists: test\\.test_view") + @Test(expectedExceptions = PrestoException.class, expectedExceptionsMessageRegExp = "View .* already exists") public void testCreateViewWithoutReplace() { SchemaTableName test = new SchemaTableName("test", "test_view"); @@ -316,13 +329,6 @@ public class TestMemoryMetadata SchemaTableName sameSchemaTableName = new SchemaTableName("test_schema", "test_renamed"); metadata.renameTable(SESSION, metadata.getTableHandle(SESSION, tableName), sameSchemaTableName); assertEquals(metadata.listTables(SESSION, Optional.of("test_schema")), ImmutableList.of(sameSchemaTableName)); - - // rename table to different schema - metadata.createSchema(SESSION, "test_different_schema", ImmutableMap.of()); - SchemaTableName differentSchemaTableName = new SchemaTableName("test_different_schema", "test_renamed"); - metadata.renameTable(SESSION, metadata.getTableHandle(SESSION, sameSchemaTableName), differentSchemaTableName); - assertEquals(metadata.listTables(SESSION, Optional.of("test_schema")), ImmutableList.of()); - assertEquals(metadata.listTables(SESSION, Optional.of("test_different_schema")), ImmutableList.of(differentSchemaTableName)); } private void assertNoTables() diff --git a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryPagesStore.java b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryPagesStore.java index 301bd2c44..22a060efc 100644 --- a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryPagesStore.java +++ b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemoryPagesStore.java @@ -27,6 +27,7 @@ import io.prestosql.spi.connector.ConnectorInsertTableHandle; import io.prestosql.spi.connector.ConnectorOutputTableHandle; import io.prestosql.spi.connector.ConnectorPageSink; import io.prestosql.spi.connector.ConnectorSession; +import io.prestosql.spi.type.testing.TestingTypeManager; import io.prestosql.testing.TestingConnectorSession; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -52,7 +53,7 @@ public class TestMemoryPagesStore @BeforeMethod public void setUp() { - pagesStore = new MemoryPagesStore(new MemoryConfig().setMaxDataPerNode(new DataSize(1, DataSize.Unit.MEGABYTE)), sorter); + pagesStore = new MemoryPagesStore(new MemoryConfig().setMaxDataPerNode(new DataSize(1, DataSize.Unit.MEGABYTE)), sorter, new TestingTypeManager()); pageSinkProvider = new MemoryPageSinkProvider(pagesStore, HostAddress.fromString("localhost:8080")); }