Issue 40: Refactor/cleanup metadata layer

build.gradle

@@ -55,7 +55,7 @@ dependencies {
     compile "com.facebook.presto:presto-record-decoder:${prestoVersion}"
     compile "com.facebook.presto:presto-spi:${prestoVersion}"
     compile "com.facebook.presto:presto-common:${prestoVersion}"
-
+    
     compile group: 'io.netty', name: 'netty-all', version:"{nettyVersion}"
 
     runtimeOnly "io.airlift:joda-to-java-time-bridge:3"

src/main/java/io/pravega/connectors/presto/PravegaConnectorConfig.java

@@ -50,6 +50,7 @@ public class PravegaConnectorConfig
      */
     private File tableDescriptionDir = new File("etc/pravega/");
 
+
     @NotNull
     public URI getControllerURI()
     {

src/main/java/io/pravega/connectors/presto/PravegaMetadata.java

@@ -104,8 +104,7 @@ public PravegaTableHandle getTableHandle(ConnectorSession session, SchemaTableNa
             return null;
         }
 
-        return new PravegaTableHandle(connectorId,
-                schemaTableName.getSchemaName(),
+        return new PravegaTableHandle(schemaTableName.getSchemaName(),
                 schemaTableName.getTableName(),
                 table.getObjectName(),
                 table.getObjectType(),

src/main/java/io/pravega/connectors/presto/PravegaSplitManager.java

@@ -116,10 +116,10 @@ private static ReaderType readerType(PravegaProperties properties)
 
     private void buildKVSplits(PravegaTableHandle pravegaTableHandle, ImmutableList.Builder<ConnectorSplit> splits)
     {
-        pravegaTableHandle.getOjectArgs().orElseThrow(() ->
+        pravegaTableHandle.getObjectArgs().orElseThrow(() ->
                 new IllegalArgumentException("no KF defined for " + pravegaTableHandle));
 
-        for (String kf : pravegaTableHandle.getOjectArgs().get()) {
+        for (String kf : pravegaTableHandle.getObjectArgs().get()) {
             PravegaSplit split =
                     new PravegaSplit(connectorId,
                             ObjectType.KV_TABLE,
@@ -130,7 +130,7 @@ private void buildKVSplits(PravegaTableHandle pravegaTableHandle, ImmutableList.
             splits.add(split);
         }
 
-        log.info("created " + pravegaTableHandle.getOjectArgs().get().size() + " kv splits");
+        log.info("created " + pravegaTableHandle.getObjectArgs().get().size() + " kv splits");
     }
 
     private void buildStreamSplits(final PravegaProperties properties,
@@ -139,7 +139,7 @@ private void buildStreamSplits(final PravegaProperties properties,
     {
         // TODO: Enable begin and end cuts to be configurable: https://github.com/pravega/pravega-sql/issues/24
         List<String> sourceStreams = multiSourceStream(pravegaTableHandle)
-                ? pravegaTableHandle.getOjectArgs().orElseThrow(
+                ? pravegaTableHandle.getObjectArgs().orElseThrow(
                         () -> new IllegalArgumentException("no args for multi source table found"))
                 : Collections.singletonList(pravegaTableHandle.getObjectName());
 

src/main/java/io/pravega/connectors/presto/PravegaStreamDescription.java

@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package io.pravega.connectors.presto;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -67,6 +66,16 @@ public PravegaStreamDescription(PravegaStreamDescription streamDescription, List
         this.event = Optional.of(event);
     }
 
+    public PravegaStreamDescription(PravegaStreamDescription streamDescription, List<PravegaStreamFieldGroup> event, List<String> objectArgs)
+    {
+        this.tableName = streamDescription.tableName;
+        this.schemaName = streamDescription.schemaName;
+        this.objectName = streamDescription.objectName;
+        this.objectType = streamDescription.objectType;
+        this.objectArgs = Optional.of(objectArgs);
+        this.event = Optional.of(event);
+    }
+
     @JsonProperty
     public Optional<String> getSchemaName()
     {

src/main/java/io/pravega/connectors/presto/PravegaStreamFieldGroup.java

@@ -33,7 +33,7 @@ public class PravegaStreamFieldGroup
     private final String dataFormat;
     private final Optional<String> dataSchema;
     private final Optional<List<PravegaStreamFieldDescription>> fields;
-    private final Optional<String> mapping;
+    private final Optional<String> mapping; // column prefix
 
     @JsonCreator
     public PravegaStreamFieldGroup(

src/main/java/io/pravega/connectors/presto/PravegaTableHandle.java

@@ -34,11 +34,6 @@
 public final class PravegaTableHandle
         implements ConnectorTableHandle
 {
-    /**
-     * connector id
-     */
-    private final String connectorId;
-
     /**
      * The schema name for this table. Is set through configuration and read
      */
@@ -56,6 +51,11 @@ public final class PravegaTableHandle
      */
     private final String objectName;
 
+    /**
+     * optional
+     * for ObjectType.STREAM, this is list of composite streams in a multi source stream
+     * for ObjectType.KV_TABLE this is list of key families
+     */
     private final Optional<List<String>> objectArgs;
 
     private final List<PravegaObjectSchema> schema;
@@ -64,7 +64,6 @@ public final class PravegaTableHandle
 
     @JsonCreator
     public PravegaTableHandle(
-            @JsonProperty("connectorId") String connectorId,
             @JsonProperty("schemaName") String schemaName,
             @JsonProperty("tableName") String tableName,
             @JsonProperty("objectName") String objectName,
@@ -73,7 +72,6 @@ public PravegaTableHandle(
             @JsonProperty("schema") List<PravegaObjectSchema> schema,
             @JsonProperty("schemaRegistryGroupId") String schemaRegistryGroupId)
     {
-        this.connectorId = requireNonNull(connectorId, "connectorId is null");
         this.schemaName = requireNonNull(schemaName, "schemaName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.objectName = requireNonNull(objectName, "objectName is null");
@@ -83,10 +81,20 @@ public PravegaTableHandle(
         this.schemaRegistryGroupId = requireNonNull(schemaRegistryGroupId, "schemaRegistryGroupId is null");
     }
 
-    @JsonProperty
-    public String getConnectorId()
+    public PravegaTableHandle(
+            @JsonProperty("schemaName") String schemaName,
+            @JsonProperty("tableName") String tableName,
+            @JsonProperty("objectName") String objectName,
+            @JsonProperty("objectType") ObjectType objectType,
+            @JsonProperty("objectArgs") Optional<List<String>> objectArgs)
     {
-        return connectorId;
+        this.schemaName = requireNonNull(schemaName, "schemaName is null");
+        this.tableName = requireNonNull(tableName, "tableName is null");
+        this.objectName = requireNonNull(objectName, "objectName is null");
+        this.objectType = requireNonNull(objectType, "objectType is null");
+        this.objectArgs = objectArgs;
+        this.schema = null;
+        this.schemaRegistryGroupId = null;
     }
 
     @JsonProperty
@@ -114,7 +122,7 @@ public String getObjectName()
     }
 
     @JsonProperty
-    public Optional<List<String>> getOjectArgs()
+    public Optional<List<String>> getObjectArgs()
     {
         return objectArgs;
     }
@@ -139,7 +147,7 @@ public SchemaTableName toSchemaTableName()
     @Override
     public int hashCode()
     {
-        return Objects.hash(connectorId, schemaName, tableName, objectName, objectType, schema);
+        return Objects.hash(schemaName, tableName, objectName, objectType, schema);
     }
 
     @Override
@@ -153,8 +161,7 @@ public boolean equals(Object obj)
         }
 
         PravegaTableHandle other = (PravegaTableHandle) obj;
-        return Objects.equals(this.connectorId, other.connectorId)
-                && Objects.equals(this.schemaName, other.schemaName)
+        return Objects.equals(this.schemaName, other.schemaName)
                 && Objects.equals(this.tableName, other.tableName)
                 && Objects.equals(this.objectName, other.objectName)
                 && Objects.equals(this.objectType, other.objectType)
@@ -165,7 +172,6 @@ public boolean equals(Object obj)
     public String toString()
     {
         return toStringHelper(this)
-                .add("connectorId", connectorId)
                 .add("schemaName", schemaName)
                 .add("tableName", tableName)
                 .add("objectName", objectName)

src/main/java/io/pravega/connectors/presto/schemamanagement/CompositeSchemaRegistry.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright (c) Pravega Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.pravega.connectors.presto.schemamanagement;
+
+import com.facebook.airlift.json.JsonCodec;
+import com.facebook.presto.spi.SchemaTableName;
+import com.google.common.annotations.VisibleForTesting;
+import io.pravega.connectors.presto.PravegaConnectorConfig;
+import io.pravega.connectors.presto.PravegaStreamDescription;
+import io.pravega.connectors.presto.PravegaStreamFieldGroup;
+import io.pravega.connectors.presto.PravegaTableHandle;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class CompositeSchemaRegistry
+        implements SchemaSupplier, SchemaRegistry {
+    private final List<SchemaSupplier> schemaSuppliers;
+
+    private final List<SchemaRegistry> schemaRegistries;
+
+    public CompositeSchemaRegistry(PravegaConnectorConfig config, JsonCodec<PravegaStreamDescription> streamDescriptionCodec) {
+        schemaSuppliers = new ArrayList<>();
+        schemaRegistries = new ArrayList<>();
+
+        // local will override, always add first
+        if (config.getTableDescriptionDir() != null &&
+                config.getTableDescriptionDir().exists() &&
+                config.getTableDescriptionDir().isDirectory()) {
+            LocalSchemaRegistry schemaRegistry =
+                    new LocalSchemaRegistry(config.getTableDescriptionDir(), streamDescriptionCodec);
+            schemaSuppliers.add(schemaRegistry);
+            schemaRegistries.add(schemaRegistry);
+        }
+
+        if (config.getSchemaRegistryURI() != null) {
+            PravegaSchemaRegistry schemaRegistry =
+                    new PravegaSchemaRegistry(config.getControllerURI(), config.getSchemaRegistryURI());
+            schemaSuppliers.add(schemaRegistry);
+            schemaRegistries.add(schemaRegistry);
+        }
+    }
+
+    @VisibleForTesting
+    public CompositeSchemaRegistry(List<SchemaSupplier> schemaSuppliers, List<SchemaRegistry> schemaRegistries)
+    {
+        this.schemaSuppliers = schemaSuppliers;
+        this.schemaRegistries = schemaRegistries;
+    }
+
+    @Override
+    public List<String> listSchemas()
+    {
+        final List<String> schemas = new ArrayList<>();
+        schemaSuppliers.forEach(p -> schemas.addAll(p.listSchemas()));
+        return schemas;
+    }
+
+    @Override
+    public List<PravegaTableHandle> listTables(String schema)
+    {
+        final List<PravegaTableHandle> tables = new ArrayList<>();
+        schemaSuppliers.forEach(p -> tables.addAll(p.listTables(schema)));
+        return tables;
+    }
+
+    @Override
+    public List<PravegaStreamFieldGroup> getSchema(SchemaTableName schemaTableName) {
+        for (SchemaRegistry schemaRegistry : schemaRegistries) {
+            List<PravegaStreamFieldGroup> schema = schemaRegistry.getSchema(schemaTableName);
+            if (schema != null) {
+                return schema;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public PravegaStreamDescription getTable(SchemaTableName schemaTableName)
+    {
+        for (SchemaRegistry schemaRegistry : schemaRegistries) {
+            PravegaStreamDescription streamDescription = schemaRegistry.getTable(schemaTableName);
+            if (streamDescription != null) {
+                return streamDescription;
+            }
+        }
+        return null;
+    }
+}