Samza SQL implementation for basic projects, filtering and UDFs

## Samza SQL implementation for basic projects, filtering and

## Design document:
https://docs.google.com/document/d/1bE-ZuPfTpntm1hT3GwQEShYDiTqU3IkxeP4-3ZcGHgU/edit?usp=sharing

Author: Srinivasulu Punuru <[email protected]>

Reviewers: Yi Pan <[email protected]>

Closes apache#295 from srinipunuru/samza-sql.1

Author: srinipunuru

build.gradle

@@ -104,6 +104,12 @@ allprojects {
   }
 }
 
+idea {
+  project {
+    languageLevel = 1.8
+  }
+}
+
 subprojects {
   apply plugin: 'eclipse'
   apply plugin: 'project-report'
@@ -185,7 +191,6 @@ project(":samza-core_$scalaVersion") {
   }
 }
 
-
 project(':samza-azure') {
   apply plugin: 'java'
   apply plugin: 'checkstyle'
@@ -215,7 +220,6 @@ project(':samza-azure') {
   }
 }
 
-
 project(":samza-autoscaling_$scalaVersion") {
   apply plugin: 'scala'
   apply plugin: 'checkstyle'
@@ -269,7 +273,24 @@ project(':samza-elasticsearch') {
     testCompile "junit:junit:$junitVersion"
     testCompile "org.mockito:mockito-all:$mockitoVersion"
 
-    // Logging in tests is good.
+    testRuntime "org.slf4j:slf4j-simple:$slf4jVersion"
+  }
+}
+
+project(':samza-sql') {
+  apply plugin: 'java'
+
+  dependencies {
+    compile project(':samza-api')
+    compile project(":samza-kafka_$scalaVersion")
+    compile "org.apache.avro:avro:$avroVersion"
+    compile "org.apache.calcite:calcite-core:$calciteVersion"
+    compile "org.slf4j:slf4j-api:$slf4jVersion"
+
+    testCompile project(":samza-test_$scalaVersion")
+    testCompile "junit:junit:$junitVersion"
+    testCompile "org.mockito:mockito-all:$mockitoVersion"
+
     testRuntime "org.slf4j:slf4j-simple:$slf4jVersion"
   }
 }
@@ -449,7 +470,7 @@ project(":samza-shell") {
   }
 
   // Usage: ./gradlew samza-shell:runJob \
-  //    -PconfigPath=file:///path/to/job/config.properties
+  //  -PconfigPath=file:///path/to/job/config.properties
   task runJob(type:JavaExec) {
     description 'To run a job (defined in a properties file)'
     main = 'org.apache.samza.job.JobRunner'
@@ -459,7 +480,7 @@ project(":samza-shell") {
   }
 
   // Usage: ./gradlew samza-shell:checkpointTool \
-  //    -PconfigPath=file:///path/to/job/config.properties -PnewOffsets=file:///path/to/new/offsets.properties
+  //  -PconfigPath=file:///path/to/job/config.properties -PnewOffsets=file:///path/to/new/offsets.properties
   task checkpointTool(type:JavaExec) {
     description 'Command-line tool to inspect and manipulate the job’s checkpoint'
     main = 'org.apache.samza.checkpoint.CheckpointTool'
@@ -470,7 +491,7 @@ project(":samza-shell") {
   }
 
   // Usage: ./gradlew samza-shell:kvPerformanceTest
-  //    -PconfigPath=file:///path/to/job/config.properties
+  //  -PconfigPath=file:///path/to/job/config.properties
   task kvPerformanceTest(type:JavaExec) {
     description 'Command-line tool to run key-value performance tests'
     main = 'org.apache.samza.test.performance.TestKeyValuePerformance'

gradle/dependency-versions.gradle

@@ -18,6 +18,8 @@
  */
  ext {
   apacheCommonsCollections4Version = "4.0"
+  avroVersion = "1.7.0"
+  calciteVersion = "1.14.0"
   commonsCodecVersion = "1.9"
   commonsCollectionVersion = "3.2.1"
   commonsHttpClientVersion = "3.1"

samza-api/src/main/java/org/apache/samza/sql/udfs/ScalarUdf.java

@@ -0,0 +1,48 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you 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 org.apache.samza.sql.udfs;
+
+import org.apache.samza.config.Config;
+
+
+/**
+ * The base class for the Scalar UDFs. All the scalar UDF classes needs to extend this and implement a method named
+ * "execute". The number and type of arguments for the execute method in the UDF class should match the number and type of fields
+ * used while invoking this UDF in SQL statement.
+ * Say for e.g. User creates a UDF class with signature int execute(int var1, String var2). It can be used in a SQL query
+ *     select myudf(id, name) from profile
+ * In the above query, Profile should contain fields named 'id' of INTEGER/NUMBER type and 'name' of type VARCHAR/CHARACTER
+ */
+public interface ScalarUdf {
+  /**
+   * Udfs can implement this method to perform any initialization that they may need.
+   * @param udfConfig Config specific to the udf.
+   */
+  void init(Config udfConfig);
+
+  /**
+   * Actual implementation of the udf function
+   * @param args
+   *   list of all arguments that the udf needs
+   * @return
+   *   Return value from the scalar udf.
+   */
+  Object execute(Object... args);
+}

samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java

@@ -0,0 +1,183 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you 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 org.apache.samza.sql.avro;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.operators.KV;
+import org.apache.samza.sql.data.SamzaSqlRelMessage;
+import org.apache.samza.sql.interfaces.SamzaRelConverter;
+import org.apache.samza.system.SystemStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class converts a Samza Avro messages to Relational messages and vice versa.
+ * This supports Samza messages where Key is a string and Value is an avro record.
+ *
+ * Conversion from Samza to Relational Message :
+ *     The key part of the samza message is represented as a special column {@link SamzaSqlRelMessage#KEY_NAME}
+ *     in relational message.
+ *
+ *     The value part of the samza message is expected to be {@link IndexedRecord}, All the fields in the IndexedRecord form
+ *     the corresponding fields of the relational message.
+ *
+ * Conversion from Relational to Samza Message :
+ *     This converts the Samza relational message into Avro {@link GenericRecord}.
+ *     All the fields of the relational message is become fields of the Avro GenericRecord except of the field with name
+ *     {@link SamzaSqlRelMessage#KEY_NAME}. This special field becomes the Key in the output Samza message.
+ */
+public class AvroRelConverter implements SamzaRelConverter {
+
+  protected final Config config;
+  private final Schema avroSchema;
+  private final RelDataType relationalSchema;
+
+  /**
+   * Class that converts the avro field to their corresponding relational fields
+   * Array fields are converted from Avro {@link org.apache.avro.generic.GenericData.Array} to {@link ArrayList}
+   */
+  public enum AvroToRelObjConverter {
+
+    /**
+     * If the relational field type is ArraySqlType, We expect the avro field to be of type either
+     * {@link GenericData.Array} or {@link List} which then is converted to Rel field of type {@link ArrayList}
+     */
+    ArraySqlType {
+      @Override
+      Object convert(Object avroObj) {
+        ArrayList<Object> retVal = new ArrayList<>();
+        if (avroObj != null) {
+          if (avroObj instanceof GenericData.Array) {
+            retVal.addAll(((GenericData.Array) avroObj));
+          } else if (avroObj instanceof List) {
+            retVal.addAll((List) avroObj);
+          }
+        }
+
+        return retVal;
+      }
+    },
+
+    /**
+     * If the relational field type is MapSqlType, We expect the avro field to be of type
+     * {@link Map}
+     */
+    MapSqlType {
+      @Override
+      Object convert(Object obj) {
+        Map<String, Object> retVal = new HashMap<>();
+        if (obj != null) {
+          retVal.putAll((Map<String, ?>) obj);
+        }
+        return retVal;
+      }
+    },
+
+    /**
+     * If the relational field type is RelRecordType, The field is considered an object
+     * and moved to rel field without any translation.
+     */
+    RelRecordType {
+      @Override
+      Object convert(Object obj) {
+        return obj;
+      }
+    },
+
+    /**
+     * If the relational field type is BasicSqlType, The field is moved to rel field without any translation.
+     */
+    BasicSqlType {
+      @Override
+      Object convert(Object obj) {
+        return obj;
+      }
+    };
+
+    abstract Object convert(Object obj);
+  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(AvroRelConverter.class);
+
+  private final Schema arraySchema = Schema.parse(
+      "{\"type\":\"array\",\"items\":{\"type\":\"record\",\"name\":\"Object\",\"namespace\":\"java.lang\",\"fields\":[]},\"java-class\":\"java.util.List\"}");
+  private final Schema mapSchema = Schema.parse(
+      "{\"type\":\"map\",\"values\":{\"type\":\"record\",\"name\":\"Object\",\"namespace\":\"java.lang\",\"fields\":[]}}");
+
+
+  public AvroRelConverter(SystemStream systemStream, AvroRelSchemaProvider schemaProvider, Config config) {
+    this.config = config;
+    this.relationalSchema = schemaProvider.getRelationalSchema();
+    this.avroSchema = Schema.parse(schemaProvider.getSchema(systemStream));
+  }
+
+  @Override
+  public SamzaSqlRelMessage convertToRelMessage(KV<Object, Object> samzaMessage) {
+    List<Object> values = new ArrayList<>();
+    List<String> fieldNames = new ArrayList<>();
+    Object value = samzaMessage.getValue();
+    if (value instanceof IndexedRecord) {
+      IndexedRecord record = (IndexedRecord) value;
+      fieldNames.addAll(relationalSchema.getFieldNames());
+      values.addAll(relationalSchema.getFieldList()
+          .stream()
+          .map(x -> getRelField(x.getType(), record.get(this.avroSchema.getField(x.getName()).pos())))
+          .collect(Collectors.toList()));
+    } else if (value == null) {
+      fieldNames.addAll(relationalSchema.getFieldNames());
+      IntStream.range(0, fieldNames.size() - 1).forEach(x -> values.add(null));
+    } else {
+      String msg = "Avro message converter doesn't support messages of type " + value.getClass();
+      LOG.error(msg);
+      throw new SamzaException(msg);
+    }
+
+    return new SamzaSqlRelMessage(samzaMessage.getKey(), fieldNames, values);
+  }
+
+  @Override
+  public KV<Object, Object> convertToSamzaMessage(SamzaSqlRelMessage relMessage) {
+    GenericRecord record = new GenericData.Record(this.avroSchema);
+    List<String> fieldNames = relMessage.getFieldNames();
+    List<Object> values = relMessage.getFieldValues();
+    for (int index = 0; index < fieldNames.size(); index++) {
+      record.put(fieldNames.get(index), values.get(index));
+    }
+
+    return new KV<>(relMessage.getKey(), record);
+  }
+
+  private Object getRelField(RelDataType relType, Object avroObj) {
+    return AvroToRelObjConverter.valueOf(relType.getClass().getSimpleName()).convert(avroObj);
+  }
+}

samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverterFactory.java

@@ -0,0 +1,44 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you 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 org.apache.samza.sql.avro;
+
+import java.util.HashMap;
+import java.util.Properties;
+import org.apache.samza.config.Config;
+import org.apache.samza.sql.interfaces.RelSchemaProvider;
+import org.apache.samza.sql.interfaces.SamzaRelConverter;
+import org.apache.samza.sql.interfaces.SamzaRelConverterFactory;
+import org.apache.samza.system.SystemStream;
+
+
+/**
+ * Avro Schema Resolver that uses static config to return a schema for a SystemStream.
+ * Schemas are configured using the config of format {systemName}.{streamName}.schema.
+ */
+public class AvroRelConverterFactory implements SamzaRelConverterFactory {
+
+  private final HashMap<SystemStream, SamzaRelConverter> relConverters = new HashMap<>();
+
+  @Override
+  public SamzaRelConverter create(SystemStream systemStream, RelSchemaProvider schemaProvider, Config config) {
+    return relConverters.computeIfAbsent(systemStream,
+        ss -> new AvroRelConverter(ss, (AvroRelSchemaProvider) schemaProvider, config));
+  }
+}

samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelSchemaProvider.java

@@ -0,0 +1,28 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you 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 org.apache.samza.sql.avro;
+
+import org.apache.samza.sql.interfaces.RelSchemaProvider;
+import org.apache.samza.system.SystemStream;
+
+
+public interface AvroRelSchemaProvider extends RelSchemaProvider {
+  String getSchema(SystemStream systemStream);
+}