Issue 10: Extend schema registry integration

build.gradle

@@ -24,6 +24,10 @@ repositories {
     maven {
         url = uri('https://repo.maven.apache.org/maven2')
     }
+
+    maven {
+        url = uri('https://packages.confluent.io/maven/')
+    }
 }
 
 dependencies {
@@ -56,6 +60,8 @@ dependencies {
     compile "com.facebook.presto:presto-spi:${prestoVersion}"
     compile "com.facebook.presto:presto-common:${prestoVersion}"
 
+    compile "io.confluent:kafka-schema-registry-client:${confluentVersion}"
+
     compile group: 'io.netty', name: 'netty-all', version:"{nettyVersion}"
 
     runtimeOnly "io.airlift:joda-to-java-time-bridge:3"

gradle.properties

@@ -18,6 +18,7 @@ airliftTestingVersion=0.191
 airliftUnitsVersion=1.3
 avroVersion=1.8.1
 commonsVersion=3.7
+confluentVersion=6.1.0
 checkstyleToolVersion=8.23
 everitJsonSchemaVersion=1.12.1
 guavaVersion=26.0-jre

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

@@ -50,6 +50,9 @@ public class PravegaConnectorConfig
      */
     private File tableDescriptionDir = new File("etc/pravega/");
 
+
+    private URI confluentSchemaRegistry;
+
     @NotNull
     public URI getControllerURI()
     {
@@ -105,4 +108,16 @@ public PravegaConnectorConfig setHideInternalColumns(boolean hideInternalColumns
         this.hideInternalColumns = hideInternalColumns;
         return this;
     }
+
+    @Config("pravega.confluentSchemaRegistry")
+    public PravegaConnectorConfig setConfluentSchemaRegistry(URI confluentSchemaRegistry)
+    {
+        this.confluentSchemaRegistry = confluentSchemaRegistry;
+        return this;
+    }
+
+    public URI getConfluentSchemaRegistry()
+    {
+        return confluentSchemaRegistry;
+    }
 }

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(