finos#1296 add type conversions required for sql filters

plugin/ignite-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/IgniteSqlFilterClause.scala

@@ -3,6 +3,7 @@ package org.finos.vuu.feature.ignite.filter
 import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.core.table.DataType.{CharDataType, StringDataType}
 import org.finos.vuu.feature.ignite.filter.IgniteSqlFilterClause.EMPTY_SQL
+import org.finos.vuu.feature.ignite.filter.SqlFilterColumnValueParser.ParsedResult
 import org.finos.vuu.util.schema.SchemaMapper
 
 private object IgniteSqlFilterClause {
@@ -29,12 +30,12 @@ case class AndIgniteSqlFilterClause(clauses:List[IgniteSqlFilterClause]) extends
 
 case class EqIgniteSqlFilterClause(columnName: String, value: String) extends IgniteSqlFilterClause {
   override def toSql(schemaMapper: SchemaMapper): String =
-    schemaMapper.externalSchemaField(columnName) match {
-      case Some(f) => f.dataType match {
-        case CharDataType | StringDataType => eqSql(f.name, quotedString(value))
-        case _ => eqSql(f.name, value)
+    SqlFilterColumnValueParser(schemaMapper).parseColumnValue(columnName, value) match {
+      case Right(ParsedResult(f, parsedValue)) => f.dataType match {
+        case CharDataType | StringDataType => eqSql(f.name, quotedString(parsedValue))
+        case _ => eqSql(f.name, parsedValue)
       }
-      case None => logMappingErrorAndReturnEmptySql(columnName)
+      case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
 
   private def eqSql(field: String, processedVal: String): String = {

plugin/ignite-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/SqlFilterColumnValueParser.scala

@@ -0,0 +1,83 @@
+package org.finos.vuu.feature.ignite.filter
+
+import com.typesafe.scalalogging.StrictLogging
+import org.finos.vuu.core.table.{Column, DataType}
+import org.finos.vuu.feature.ignite.filter.SqlFilterColumnValueParser.{ErrorMessage, ParsedResult}
+import org.finos.vuu.util.schema.typeConversion.TypeConverter.buildConverterName
+import org.finos.vuu.util.schema.{SchemaField, SchemaMapper}
+
+protected trait SqlFilterColumnValueParser {
+  def parseColumnValue(columnName: String, columnValue: String): Either[ErrorMessage, ParsedResult[String]]
+  def parseColumnValues(columnName: String, columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[String]]]
+}
+
+protected object SqlFilterColumnValueParser {
+  def apply(schemaMapper: SchemaMapper): SqlFilterColumnValueParser = new ColumnValueParser(schemaMapper)
+
+  case class ParsedResult[T](externalField: SchemaField, data: T)
+
+  type ErrorMessage = String
+}
+
+private class ColumnValueParser(private val mapper: SchemaMapper) extends SqlFilterColumnValueParser with StrictLogging {
+
+  override def parseColumnValue(columnName: String, columnValue: String): Either[ErrorMessage, ParsedResult[String]] = {
+    mapper.externalSchemaField(columnName) match {
+      case Some(f) => RawColumnValueParser(f).parse(columnValue).map(ParsedResult(f, _))
+      case None    => Left(externalFieldNotFoundError(columnName))
+    }
+  }
+
+  override def parseColumnValues(columnName: String, columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[String]]] = {
+    mapper.externalSchemaField(columnName) match {
+      case Some(f) => parseColumnValues(RawColumnValueParser(f), columnValues)
+      case None    => Left(externalFieldNotFoundError(columnName))
+    }
+  }
+
+  private def parseColumnValues(parser: RawColumnValueParser,
+                                columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[String]]] = {
+    val (errors, parsedValues) = columnValues.partitionMap(parser.parse)
+    val combinedError = errors.mkString("\n")
+
+    if (parsedValues.isEmpty) return Left(combinedError)
+
+    if (errors.nonEmpty) logger.error(s"Failed to parse some of the column values corresponding to " +
+      s"the column ${parser.column.name}: \n $combinedError"
+    )
+
+    Right(ParsedResult(parser.field, parsedValues))
+  }
+
+  private def externalFieldNotFoundError(columnName: String): String =
+    s"Failed to find mapped external field for column `$columnName`"
+
+  private case class RawColumnValueParser(field: SchemaField) {
+    val column: Column = mapper.tableColumn(field.name).get
+
+    def parse(columnValue: String): Either[ErrorMessage, String] = {
+      parseStringToColumnDataType(columnValue)
+        .flatMap(convertColumnValueToExternalFieldType)
+        .map(convertExternalValueToString)
+    }
+
+    private def parseStringToColumnDataType(value: String): Either[ErrorMessage, Any] =
+      DataType.parseDataType(value, column.dataType)
+
+    private def convertColumnValueToExternalFieldType(columnValue: Any): Either[ErrorMessage, Any] =
+      mapper.toMappedExternalFieldType(columnValue, column.name)
+        .toRight(s"Failed to convert column value `$columnValue` from `${column.dataType}` to external type `${field.dataType}`")
+
+    private def convertExternalValueToString(v: Any): String =
+      mapper.convertExternalValueToString(v, field.name).getOrElse(logWarningAndUseDefaultToString(v))
+
+    private def logWarningAndUseDefaultToString(value: Any): String = {
+      logger.warn(
+        s"Could not find a converter [${buildConverterName(field.dataType, classOf[String])}] " +
+          s"required for SQL filters to be applied to ${field.name}. Falling back to using the " +
+          s"default `toString`."
+      )
+      Option(value).map(_.toString).orNull
+    }
+  }
+}

vuu/src/main/scala/org/finos/vuu/core/table/Column.scala

@@ -2,6 +2,7 @@ package org.finos.vuu.core.table
 
 import org.finos.vuu.api.TableDef
 import org.finos.vuu.core.table.column.CalculatedColumnClause
+import org.finos.vuu.util.schema.typeConversion.{DefaultTypeConverters, TypeConverterContainerBuilder}
 
 object DataType {
 
@@ -34,6 +35,22 @@ object DataType {
     }
   }
 
+  def parseDataType[T](value: String, t: Class[T]): Either[String, T] = {
+    typeConverterContainer.convert[String, T](value, classOf[String], t) match {
+      case Some(parsedValue) => Right(parsedValue)
+      case None              => Left(s"Failed to parse String $value to data type $t")
+    }
+  }
+
+  private val typeConverterContainer = TypeConverterContainerBuilder()
+    .withoutDefaults()
+    .withConverter(DefaultTypeConverters.stringToCharConverter)
+    .withConverter(DefaultTypeConverters.stringToBooleanConverter)
+    .withConverter(DefaultTypeConverters.stringToIntConverter)
+    .withConverter(DefaultTypeConverters.stringToLongConverter)
+    .withConverter(DefaultTypeConverters.stringToDoubleConverter)
+    .build()
+
 }
 
 object Columns {

vuu/src/main/scala/org/finos/vuu/util/schema/SchemaMapper.scala

@@ -15,6 +15,9 @@ import org.finos.vuu.util.schema.typeConversion.{TypeConverter, TypeConverterCon
 trait SchemaMapper {
   def tableColumn(extFieldName: String): Option[Column]
   def externalSchemaField(columnName: String): Option[SchemaField]
+  def toMappedExternalFieldType(columnValue: Any, columnName: String): Option[Any]
+  def toMappedInternalColumnType(extFieldValue: Any, extFieldName: String): Option[Any]
+  def convertExternalValueToString(externalValue: Any, extFieldName: String): Option[String]
   def toInternalRowMap(externalValues: List[_]): Map[String, Any]
   def toInternalRowMap(externalDto: Product): Map[String, Any]
 }
@@ -133,22 +136,40 @@ private class SchemaMapperImpl(private val externalSchema: ExternalEntitySchema,
                                private val typeConverterContainer: TypeConverterContainer) extends SchemaMapper {
   private val externalFieldByColumnName: Map[String, SchemaField] = getExternalSchemaFieldsByColumnName
   private val internalColumnByExtFieldName: Map[String, Column] = getTableColumnByExternalField
+  private val extFieldsMap: Map[String, SchemaField] = externalSchema.fields.map(f => (f.name, f)).toMap
 
   override def tableColumn(extFieldName: String): Option[Column] = internalColumnByExtFieldName.get(extFieldName)
   override def externalSchemaField(columnName: String): Option[SchemaField] = externalFieldByColumnName.get(columnName)
+
   override def toInternalRowMap(externalValues: List[_]): Map[String, Any] = toInternalRowMap(externalValues.toArray)
   override def toInternalRowMap(externalDto: Product): Map[String, Any] = toInternalRowMap(externalDto.productIterator.toArray)
-
   private def toInternalRowMap(externalValues: Array[_]): Map[String, Any] = {
-    externalFieldByColumnName.map({ case (_, extField) =>
+    externalFieldByColumnName.map({ case (columnName, extField) =>
       val extFieldValue = externalValues(extField.index)
-      val extFieldDataType = extField.dataType.asInstanceOf[Class[Any]]
-      val column = tableColumn(extField.name).get
-      val columnValue = typeConverterContainer.convert(extFieldValue, extFieldDataType, column.dataType).get
-      (column.name, columnValue)
+      val columnValue = toMappedInternalColumnType(extFieldValue, extField.name).get
+      (columnName, columnValue)
+    })
+  }
+
+  override def toMappedExternalFieldType(columnValue: Any, columnName: String): Option[Any] = {
+    externalSchemaField(columnName).map(field => {
+      val col = tableColumn(field.name).get
+      typeConverterContainer.convert(columnValue, col.dataType.asInstanceOf[Class[Any]], field.dataType)
     })
   }
 
+  override def toMappedInternalColumnType(extFieldValue: Any, extFieldName: String): Option[Any] = {
+    tableColumn(extFieldName).map(col => {
+      val field = externalSchemaField(col.name).get
+      typeConverterContainer.convert(extFieldValue, field.dataType.asInstanceOf[Class[Any]], col.dataType)
+    })
+  }
+
+  override def convertExternalValueToString(externalValue: Any, extFieldName: String): Option[String] = {
+    extFieldsMap.get(extFieldName)
+      .flatMap(f => typeConverterContainer.convert(externalValue, f.dataType.asInstanceOf[Class[Any]], classOf[String]))
+  }
+
   private def getExternalSchemaFieldsByColumnName =
     externalSchema.fields.flatMap(f =>
       Option.when(columnNameByExternalField.contains(f.name))(columnNameByExternalField(f.name), f)

vuu/src/main/scala/org/finos/vuu/util/schema/typeConversion/DefaultTypeConverters.scala

@@ -6,6 +6,9 @@ object DefaultTypeConverters {
   val stringToDoubleConverter: TypeConverter[String, Double] = TypeConverter(classOf[String], classOf[Double], withNullSafety[String, Double](_, _.toDouble))
   val stringToLongConverter: TypeConverter[String, Long] = TypeConverter[String, Long](classOf[String], classOf[Long], withNullSafety[String, Long](_, _.toLong))
   val stringToIntConverter: TypeConverter[String, Integer] = TypeConverter(classOf[String], classOf[Integer], withNullSafety[String, Integer](_, _.toInt))
+  val stringToCharConverter: TypeConverter[String, Character] = TypeConverter(classOf[String], classOf[Character], withNullSafety[String, Character](_, _.toCharArray.apply(0)))
+  val stringToBooleanConverter: TypeConverter[String, Boolean] = TypeConverter(classOf[String], classOf[Boolean], withNullSafety[String, Boolean](_, _.toBoolean))
+
   val intToStringConverter: TypeConverter[Integer, String] = TypeConverter(classOf[Integer], classOf[String], withNullSafety[Integer, String](_, _.toString))
   val longToStringConverter: TypeConverter[Long, String] = TypeConverter(classOf[Long], classOf[String], withNullSafety[Long, String](_, _.toString))
   val doubleToStringConverter: TypeConverter[Double, String] = TypeConverter(classOf[Double], classOf[String], withNullSafety[Double, String](_, _.toString))
@@ -20,5 +23,7 @@ object DefaultTypeConverters {
     longToStringConverter,
     stringToIntConverter,
     intToStringConverter,
+    stringToCharConverter,
+    stringToBooleanConverter,
   )
 }

vuu/src/main/scala/org/finos/vuu/util/schema/typeConversion/TypeConverterContainer.scala

@@ -1,5 +1,7 @@
 package org.finos.vuu.util.schema.typeConversion
 
+import scala.util.Try
+
 trait TypeConverterContainer {
   def convert[From, To](value: From, fromClass: Class[From], toClass: Class[To]): Option[To]
   def typeConverter[From, To](name: String): Option[TypeConverter[From, To]]
@@ -13,9 +15,9 @@ private case class TypeConverterContainerImpl(
 
   override def convert[From, To](value: From, fromClass: Class[From], toClass: Class[To]): Option[To] = {
     if (TypeUtils.areTypesEqual(fromClass, toClass)) {
-      return Option(value.asInstanceOf[To])
+      return Some(value.asInstanceOf[To])
     }
-    typeConverter[From, To](fromClass, toClass).map(_.convert(value))
+    typeConverter[From, To](fromClass, toClass).flatMap(tc => Try(tc.convert(value)).toOption)
   }
 
   override def typeConverter[From, To](name: String): Option[TypeConverter[From, To]] = {