Memory Connector: Integrate with HetuMetastore

presto-memory/pom.xml

@@ -107,6 +107,12 @@
         </dependency>
 
         <!-- for testing -->
+        <dependency>
+            <groupId>io.hetu.core</groupId>
+            <artifactId>hetu-common</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.hetu.core</groupId>
             <artifactId>presto-tests</artifactId>
@@ -161,5 +167,17 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>io.hetu.core</groupId>
+            <artifactId>hetu-metastore</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.hetu.core</groupId>
+            <artifactId>hetu-filesystem-client</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

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("handle") MemoryColumnHandle handle,
-            @JsonProperty("name") String name,
+            @JsonProperty("name") String name)
-            @JsonProperty("type") Type type)
     {
         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(TypeManager typeManager)
-    public Type getType()
     {
-        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();
     }
 }

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

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

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 io.airlift.json.JsonCodec;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Maps;
 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 io.airlift.json.JsonCodec.jsonCodec;
-import static com.google.common.collect.ImmutableList.toImmutableList;
 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<ConnectorViewDefinition> VIEW_CODEC = jsonCodec(ConnectorViewDefinition.class);
 
     private final NodeManager nodeManager;
-    private final List<String> schemas = new ArrayList<>();
+    private final TypeManager typeManager;
-    private final AtomicLong nextTableId = new AtomicLong();
+    private final AtomicLong nextTableId;
-    private final Map<SchemaTableName, Long> tableIds = new ConcurrentHashMap<>();
     private final Map<Long, TableInfo> tables = new ConcurrentHashMap<>();
     private final Map<SchemaTableName, ConnectorViewDefinition> 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<CatalogEntity> 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<String> listSchemaNames(ConnectorSession session)
     {
-        return ImmutableList.copyOf(schemas);
+        ImmutableList.Builder<String> 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<String, Object> properties)
     {
-        if (schemas.contains(schemaName)) {
+        checkSchemaNotExists(schemaName);
-            throw new PrestoException(ALREADY_EXISTS, format("Schema [%s] already exists", schemaName));
+
-        }
+        DatabaseEntity.Builder databaseBuilder = DatabaseEntity.builder()
-        schemas.add(schemaName);
+                .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)) {
+        checkSchemaExists(schemaName);
-            throw new PrestoException(NOT_FOUND, format("Schema [%s] does not exist", schemaName));
-        }
 
-        boolean tablesExist = tables.values().stream()
+        if (!metastore.getAllTables(MEM_KEY, schemaName).isEmpty()) {
-                .anyMatch(table -> table.getSchemaName().equals(schemaName));
-
-        if (tablesExist) {
             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);
+        Optional<TableEntity> tableEntity = metastore.getTable(MEM_KEY, schemaTableName.getSchemaName(), schemaTableName.getTableName());
-        if (id == null) {
+        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<SchemaTableName> listTables(ConnectorSession session, Optional<String> schemaName)
     {
-        return tables.values().stream()
+        return getTableStream(schemaName, false)
-                .filter(table -> schemaName.map(table.getSchemaName()::equals).orElse(true))
+                .collect(Collectors.toList());
-                .map(TableInfo::getSchemaTableName)
-                .collect(toList());
     }
 
     @Override
     public Map<String, ColumnHandle> 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<SchemaTableName, List<ColumnMetadata>> 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());
+        TableInfo info = getTableInfo(handle.getId());
-        if (info != null) {
+        metastore.dropTable(MEM_KEY, info.getSchemaName(), info.getTableName());
-            tableIds.remove(info.getSchemaTableName());
+        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);
+        TableInfo oldInfo = getTableInfo(tableId);
-        tables.put(tableId, new TableInfo(tableId, newTableName.getSchemaName(), newTableName.getTableName(), oldInfo.getColumns(), oldInfo.getDataFragments()));
+        updateTableInfo(tableId, new TableInfo(
+                tableId,
+                newTableName.getSchemaName(),
+                newTableName.getTableName(),
+                oldInfo.getColumns(),
+                oldInfo.getDataFragments()));
 
-        tableIds.remove(oldInfo.getSchemaTableName());
+        metastore.alterTable(MEM_KEY, oldInfo.getSchemaName(), oldInfo.getTableName(),
-        tableIds.put(newTableName, tableId);
+                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<ConnectorNewTableLayout> layout)
     {
         checkSchemaExists(tableMetadata.getTable().getSchemaName());
-        checkTableNotExists(tableMetadata.getTable());
+        checkTableNotExists(tableMetadata.getTable(), false);
 
         List<SortingColumn> sortedBy = MemoryTableProperties.getSortedBy(tableMetadata.getProperties());
         if (sortedBy == null) {
@@ -250,7 +291,7 @@ public class MemoryMetadata
         Set<String> 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<Node> nodes = nodeManager.getRequiredWorkerNodes();
         checkState(!nodes.isEmpty(), "No Memory nodes available");
 
         long tableId = nextId;
 
         List<ColumnInfo> columnInfos = columns.build();
-        tableIds.put(tableMetadata.getTable(), tableId);
+        metastore.createTable(TableEntity.builder()
-        tables.put(tableId, new TableInfo(
+                .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> 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)) {
+        if (metastore.getTable(MEM_KEY, tableName.getSchemaName(), tableName.getTableName()).isPresent()) {
-            throw new PrestoException(ALREADY_EXISTS, format("Table [%s] already exists", tableName.toString()));
+            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> 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) {
+        if (!replace) {
-            views.put(viewName, definition);
+            checkTableNotExists(viewName, true);
-        }
-        else if (views.putIfAbsent(viewName, definition) != null) {
-            throw new PrestoException(ALREADY_EXISTS, "View already exists: " + viewName);
         }
+        updateViewDef(viewName, definition);
     }
 
     @Override
     public void dropView(ConnectorSession session, SchemaTableName viewName)
     {
-        if (views.remove(viewName) == null) {
+        checkTableExists(viewName, true);
-            throw new ViewNotFoundException(viewName);
+        updateViewDef(viewName, null);
-        }
     }
 
     @Override
     public List<SchemaTableName> listViews(ConnectorSession session, Optional<String> schemaName)
     {
-        return views.keySet().stream()
+        return getTableStream(schemaName, true)
-                .filter(viewName -> schemaName.map(viewName.getSchemaName()::equals).orElse(true))
+                .collect(Collectors.toList());
-                .collect(toImmutableList());
     }
 
     @Override
     public Map<SchemaTableName, ConnectorViewDefinition> getViews(ConnectorSession session, Optional<String> 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<ConnectorViewDefinition> getView(ConnectorSession session, SchemaTableName viewName)
     {
-        return Optional.ofNullable(views.get(viewName));
+        return Optional.ofNullable(getViewDef(viewName));
     }
 
     private void updateRowsOnHosts(long tableId, Collection<Slice> 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<MemoryDataFragment> 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<SchemaTableName> getTableStream(Optional<String> schemaName, boolean isView)
+    {
+        Stream<TableEntity> 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> 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> 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<Long> getTableIdSet()
+    {
+        return getMemoryCatalogEntity().getParameters().keySet().stream().filter(e -> !NEXT_ID_KEY.equals(e)).map(Long::valueOf).collect(Collectors.toSet());
+    }
 }

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

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<TableInfo> TABLE_INFO_JSON_CODEC = jsonCodec(TableInfo.class);
+
     private final long id;
     private final String schemaName;
     private final String tableName;
     private final List<ColumnInfo> columns;
     private final Map<HostAddress, MemoryDataFragment> dataFragments;
 
-    public TableInfo(long id, String schemaName, String tableName, List<ColumnInfo> columns, Map<HostAddress, MemoryDataFragment> dataFragments)
+    @JsonCreator
+    public TableInfo(
+            @JsonProperty("id") long id,
+            @JsonProperty("schemaName") String schemaName,
+            @JsonProperty("tableName") String tableName,
+            @JsonProperty("columns") List<ColumnInfo> columns,
+            @JsonProperty("dataFragments") Map<HostAddress, MemoryDataFragment> 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<ColumnInfo> getColumns()
     {
         return columns;
@@ -85,8 +103,19 @@ public class TableInfo
                 .get();
     }
 
+    @JsonProperty
     public Map<HostAddress, MemoryDataFragment> 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));
+    }
 }

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<Integer, BloomFilter> indexChannelFilters = new HashMap<>();
     private final Map<Integer, Map.Entry<Comparable, Comparable>> columnMinMax = new HashMap<>();
 
-    public LogicalPart(List<ColumnInfo> columns, List<SortingColumn> sortedBy, List<String> indexColumns, PageSorter pageSorter, long maxLogicalPartBytes)
+    public LogicalPart(List<ColumnInfo> columns, List<SortingColumn> sortedBy, List<String> 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<>();

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<Long, TableData> 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<ColumnInfo> columns, List<SortingColumn> sortedBy, List<String> 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));
         }
     }
 

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<ColumnInfo> columns, List<SortingColumn> sortedBy,
-                     List<String> indexColumns, PageSorter pageSorter, MemoryConfig config)
+            List<String> 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<LogicalPart> 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);

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<SchemaTableName> tables = metadata.listTables(SESSION, Optional.empty());
-        assertTrue(tables.size() == 1, "Expected only one table");
+        assertEquals(tables.size(), 1, "Expected only one table");
-        assertTrue(tables.get(0).getTableName().equals("temp_table"), "Expected table with name 'temp_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()

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