finos#1296 Make FilterColumnValueParser SQL independent

Also makes it responsible only for one thing i.e. parse column value (String) to external value's data type. converting external value to String is use-case specific i.e. using it in sql query, hence moved it to IgniteSqlFilterClause.
junaidzm13 · Apr 29, 2024 · 8998b9c · 8998b9c
1 parent 019abd6
commit 8998b9c
Show file tree

Hide file tree

Showing 4 changed files with 154 additions and 169 deletions.
diff --git a/...e-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/FilterColumnValueParser.scala b/...e-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/FilterColumnValueParser.scala
@@ -0,0 +1,73 @@
+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.FilterColumnValueParser.{ErrorMessage, ParsedResult}
+import org.finos.vuu.util.schema.{SchemaField, SchemaMapper}
+
+protected trait FilterColumnValueParser {
+  def parse(columnName: String, columnValue: String): Either[ErrorMessage, ParsedResult[Any]]
+  def parse(columnName: String, columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[Any]]]
+}
+
+protected object FilterColumnValueParser {
+  def apply(schemaMapper: SchemaMapper): FilterColumnValueParser = {
+    new ColumnValueParser(schemaMapper)
+  }
+
+  case class ParsedResult[T](externalField: SchemaField, externalData: T)
+
+  type ErrorMessage = String
+
+  val STRING_DATA_TYPE: Class[String] = classOf[String]
+}
+
+private class ColumnValueParser(private val mapper: SchemaMapper) extends FilterColumnValueParser with StrictLogging {
+
+  override def parse(columnName: String, columnValue: String): Either[ErrorMessage, ParsedResult[Any]] = {
+    mapper.externalSchemaField(columnName) match {
+      case Some(f) => RawColumnValueParser(f).parse(columnValue).map(ParsedResult(f, _))
+      case None    => Left(externalFieldNotFoundError(columnName))
+    }
+  }
+
+  override def parse(columnName: String, columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[Any]]] = {
+    mapper.externalSchemaField(columnName) match {
+      case Some(f) => parseValues(RawColumnValueParser(f), columnValues)
+      case None    => Left(externalFieldNotFoundError(columnName))
+    }
+  }
+
+  private def parseValues(parser: RawColumnValueParser,
+                          columnValues: List[String]): Either[ErrorMessage, ParsedResult[List[Any]]] = {
+    val (errors, parsedValues) = columnValues.partitionMap(parser.parse)
+    val combinedError = errors.mkString("\n")
+
+    if (parsedValues.isEmpty) {
+      Left(combinedError)
+    } else {
+      if (errors.nonEmpty) logger.error(
+        s"Failed to parse some of the column values corresponding to 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, Any] = {
+      parseStringToColumnDataType(columnValue).flatMap(convertColumnValueToExternalFieldType)
+    }
+
+    private def parseStringToColumnDataType(value: String): Either[ErrorMessage, Any] =
+      DataType.parseToDataType(value, column.dataType)
+
+    private def convertColumnValueToExternalFieldType(columnValue: Any): Either[ErrorMessage, Any] =
+      mapper.toMappedExternalFieldType(column.name, columnValue)
+        .toRight(s"Failed to convert column value `$columnValue` from `${column.dataType}` to external type `${field.dataType}`")
+  }
+}
diff --git a/...ite-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/IgniteSqlFilterClause.scala b/...ite-plugin/src/main/scala/org/finos/vuu/feature/ignite/filter/IgniteSqlFilterClause.scala
@@ -2,8 +2,9 @@ package org.finos.vuu.feature.ignite.filter
 
 import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.feature.ignite.filter.IgniteSqlFilterClause.EMPTY_SQL
-import org.finos.vuu.feature.ignite.filter.SqlFilterColumnValueParser.{ParsedResult, STRING_DATA_TYPE}
+import org.finos.vuu.feature.ignite.filter.FilterColumnValueParser.{ParsedResult, STRING_DATA_TYPE}
 import org.finos.vuu.util.schema.{SchemaField, SchemaMapper}
+import org.finos.vuu.util.types.TypeUtils
 
 private object IgniteSqlFilterClause {
   val EMPTY_SQL = ""
@@ -28,44 +29,55 @@ case class AndIgniteSqlFilterClause(clauses:List[IgniteSqlFilterClause]) extends
 }
 
 case class EqIgniteSqlFilterClause(columnName: String, value: String) extends IgniteSqlFilterClause {
-  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String =
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValue(columnName, value) match {
-      case Right(ParsedResult(f, parsedValue)) => eqSql(f.name, parsedValue)
+  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, value) match {
+      case Right(ParsedResult(f, externalValue)) => eqSql(f.name, toString(externalValue, f.dataType))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
+  }
 
   private def eqSql(field: String, processedVal: String): String = {
     s"$field = $processedVal"
   }
 }
 
 case class NeqIgniteSqlFilterClause(columnName: String, value: String) extends IgniteSqlFilterClause {
-  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String =
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValue(columnName, value) match {
-      case Right(ParsedResult(f, parsedValue)) => neqSql(f.name, parsedValue)
+  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, value) match {
+      case Right(ParsedResult(f, externalValue)) => neqSql(f.name, toString(externalValue, f.dataType))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
+  }
 
   private def neqSql(field: String, processedVal: String): String = {
     s"$field != $processedVal"
   }
 }
 
 case class RangeIgniteSqlFilterClause(op: RangeOp)(columnName: String, value: String) extends IgniteSqlFilterClause {
-  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String =
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValue(columnName, value) match {
-      case Right(ParsedResult(f, parsedValue)) => rangeSql(f.name, parsedValue)
+  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, value) match {
+      case Right(ParsedResult(f, externalValue)) => rangeSql(f.name, toString(externalValue, f.dataType))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
+  }
 
   private def rangeSql(field: String, processedVal: String): String = s"$field ${op.value} $processedVal"
   override def toString = s"RangeIgniteSqlFilterClause[$op]($columnName, $value)"
 }
 
 case class StartsIgniteSqlFilterClause(columnName: String, value: String) extends IgniteSqlFilterClause with StrictLogging {
   override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValue(columnName, value) match {
-      case Right(ParsedResult(f, parsedValue)) => startsSql(f, parsedValue)
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, value) match {
+      case Right(ParsedResult(f, externalValue)) => startsSql(f, toString(externalValue, f.dataType))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
   }
@@ -78,8 +90,10 @@ case class StartsIgniteSqlFilterClause(columnName: String, value: String) extend
 
 case class EndsIgniteSqlFilterClause(columnName: String, value: String) extends IgniteSqlFilterClause with StrictLogging {
   override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValue(columnName, value) match {
-      case Right(ParsedResult(f, parsedValue)) => endsSql(f, parsedValue)
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, value) match {
+      case Right(ParsedResult(f, externalValue)) => endsSql(f, toString(externalValue, f.dataType))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
   }
@@ -91,11 +105,14 @@ case class EndsIgniteSqlFilterClause(columnName: String, value: String) extends
 }
 
 case class InIgniteSqlFilterClause(columnName: String, values: List[String]) extends IgniteSqlFilterClause with StrictLogging {
-  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String =
-    SqlFilterColumnValueParser(schemaMapper, toStringContainer).parseColumnValues(columnName, values) match {
-      case Right(ParsedResult(f, parsedValues)) => inQuery(f.name, parsedValues)
+  override def toSql(schemaMapper: SchemaMapper, toStringContainer: SqlStringConverterContainer): String = {
+    def toString = getStringConverter(toStringContainer)
+
+    FilterColumnValueParser(schemaMapper).parse(columnName, values) match {
+      case Right(ParsedResult(f, externalValues)) => inQuery(f.name, externalValues.map(toString(_, f.dataType)))
       case Left(errMsg) => logErrorAndReturnEmptySql(errMsg)
     }
+  }
 
   private def inQuery(field: String, processedValues: List[String]) = {
     s"$field IN (${processedValues.mkString(",")})"
@@ -110,6 +127,37 @@ object RangeOp {
   final case object LTE extends RangeOp(value = "<=")
 }
 
+
+private object getStringConverter {
+  def apply(toStringContainer: SqlStringConverterContainer): (Any, Class[_]) => String = (value, dataType) =>
+      if (TypeUtils.areTypesEqual(dataType, STRING_DATA_TYPE)) {
+        quotedString(defaultToString(value))
+      } else {
+        toStringContainer
+          .toString(value, dataType.asInstanceOf[Class[Any]])
+          .getOrElse(addQuotesIfRequired(defaultToString(value), dataType))
+      }
+
+    private def defaultToString(value: Any): String = Option(value).map(_.toString).orNull
+}
+
+private object quotedString {
+  def apply(s: String) = s"'$s'"
+}
+
+private object addQuotesIfRequired {
+  def apply(v: String, dataType: Class[_]): String = if (requireQuotes(dataType)) quotedString(v) else v
+
+  private def requireQuotes(dt: Class[_]): Boolean = dataTypesRequiringQuotes.contains(dt)
+
+  private val dataTypesRequiringQuotes: Set[Class[_]] = Set(
+    classOf[String],
+    classOf[Char],
+    classOf[java.lang.Character],
+    classOf[java.sql.Date],
+  )
+}
+
 private object logErrorAndReturnEmptySql extends StrictLogging {
   def apply(error: String): String = {
     logger.error(error)

diff --git a/...lugin/src/main/scala/org/finos/vuu/feature/ignite/filter/SqlFilterColumnValueParser.scala b/...lugin/src/main/scala/org/finos/vuu/feature/ignite/filter/SqlFilterColumnValueParser.scala