From 067cc4270d84456443b46a72b0e5797488aa38c4 Mon Sep 17 00:00:00 2001 From: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com> Date: Thu, 28 Nov 2024 22:54:24 -0500 Subject: [PATCH 1/2] apply fix --- .../amber/core/tuple/AttributeTypeUtils.scala | 228 +++++++++++------- .../core/tuple/AttributeTypeUtilsSpec.scala | 92 ++++++- 2 files changed, 237 insertions(+), 83 deletions(-) diff --git a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala index 8cb7934011a..12f21cb6597 100644 --- a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala +++ b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala @@ -3,6 +3,8 @@ package edu.uci.ics.amber.core.tuple import com.github.sisyphsu.dateparser.DateParserUtils import java.sql.Timestamp +import java.text.NumberFormat +import java.util.Locale import scala.util.Try import scala.util.control.Exception.allCatch @@ -19,10 +21,10 @@ object AttributeTypeUtils extends Serializable { * @return schema of data */ def SchemaCasting( - schema: Schema, - attribute: String, - resultType: AttributeType - ): Schema = { + schema: Schema, + attribute: String, + resultType: AttributeType + ): Schema = { // need a builder to maintain the order of original schema val builder = Schema.builder() val attributes: List[Attribute] = schema.getAttributes @@ -31,8 +33,8 @@ object AttributeTypeUtils extends Serializable { if (attributes.apply(i).getName.equals(attribute)) { resultType match { case AttributeType.STRING | AttributeType.INTEGER | AttributeType.DOUBLE | - AttributeType.LONG | AttributeType.BOOLEAN | AttributeType.TIMESTAMP | - AttributeType.BINARY => + AttributeType.LONG | AttributeType.BOOLEAN | AttributeType.TIMESTAMP | + AttributeType.BINARY => builder.add(attribute, resultType) case AttributeType.ANY | _ => builder.add(attribute, attributes.apply(i).getType) @@ -58,13 +60,13 @@ object AttributeTypeUtils extends Serializable { * as specified by the typeCastingUnits. */ def tupleCasting( - tuple: Tuple, - targetTypes: Map[String, AttributeType] - ): TupleLike = + tuple: Tuple, + targetTypes: Map[String, AttributeType] + ): TupleLike = TupleLike( tuple.getSchema.getAttributes.map { attr => val targetType = targetTypes.getOrElse(attr.getName, attr.getType) - parseField(tuple.getField(attr.getName), targetType) + parseField(tuple.getField(attr.getName), targetType, force = true) } ) @@ -80,9 +82,9 @@ object AttributeTypeUtils extends Serializable { */ @throws[AttributeTypeException] def parseFields( - fields: Array[Any], - attributeTypes: Array[AttributeType] - ): Array[Any] = { + fields: Array[Any], + attributeTypes: Array[AttributeType] + ): Array[Any] = { fields.indices.map(i => parseField(fields(i), attributeTypes(i))).toArray } @@ -90,18 +92,21 @@ object AttributeTypeUtils extends Serializable { * parse Field to a corresponding Java object base on the given Schema AttributeType * @param field fields value * @param attributeType target AttributeType + * @param force force to parse the field to the target type if possible + * currently only support for comma-separated numbers * * @return parsedField in the target AttributeType */ @throws[AttributeTypeException] def parseField( - field: Any, - attributeType: AttributeType - ): Any = { + field: Any, + attributeType: AttributeType, + force: Boolean = false + ): Any = { if (field == null) return null attributeType match { - case AttributeType.INTEGER => parseInteger(field) - case AttributeType.LONG => parseLong(field) + case AttributeType.INTEGER => parseInteger(field, force) + case AttributeType.LONG => parseLong(field, force) case AttributeType.DOUBLE => parseDouble(field) case AttributeType.BOOLEAN => parseBoolean(field) case AttributeType.TIMESTAMP => parseTimestamp(field) @@ -112,87 +117,147 @@ object AttributeTypeUtils extends Serializable { } @throws[AttributeTypeException] - private def parseInteger(fieldValue: Any): Integer = { - fieldValue match { - case str: String => str.trim.toInt - case int: Integer => int - case long: java.lang.Long => long.toInt - case double: java.lang.Double => double.toInt - case boolean: java.lang.Boolean => if (boolean) 1 else 0 - // Timestamp and Binary are considered to be illegal here. - case _ => + private def parseInteger(fieldValue: Any, force: Boolean = false): Integer = { + val attempt: Try[Integer] = Try { + fieldValue match { + case str: String => + if (force) { + // Use US locale for comma-separated numbers + NumberFormat.getNumberInstance(Locale.US).parse(str.trim).intValue() + } else { + str.trim.toInt + } + case int: Integer => int + case long: java.lang.Long => long.toInt + case double: java.lang.Double => double.toInt + case boolean: java.lang.Boolean => if (boolean) 1 else 0 + // Timestamp and Binary are considered to be illegal here. + case _ => + throw new IllegalArgumentException( + s"Unsupported type for parsing to Integer: ${fieldValue.getClass.getName}" + ) + } + } + + attempt.recover { + case e: Exception => throw new AttributeTypeException( - s"not able to parse type ${fieldValue.getClass} to Integer: ${fieldValue.toString}" + s"Failed to parse type ${fieldValue.getClass.getName} to Integer: ${fieldValue.toString}", + e ) - } + }.get } @throws[AttributeTypeException] - private def parseLong(fieldValue: Any): java.lang.Long = { - fieldValue match { - case str: String => str.trim.toLong - case int: Integer => int.toLong - case long: java.lang.Long => long - case double: java.lang.Double => double.toLong - case boolean: java.lang.Boolean => if (boolean) 1L else 0L - case timestamp: Timestamp => timestamp.toInstant.toEpochMilli - // Binary is considered to be illegal here. - case _ => + private def parseLong(fieldValue: Any, force: Boolean = false): java.lang.Long = { + val attempt: Try[Long] = Try { + fieldValue match { + case str: String => + if (force) { + // Use US locale for comma-separated numbers + NumberFormat.getNumberInstance(Locale.US).parse(str.trim).longValue() + } else { + str.trim.toLong + } + case int: Integer => int.toLong + case long: java.lang.Long => long + case double: java.lang.Double => double.toLong + case boolean: java.lang.Boolean => if (boolean) 1L else 0L + case timestamp: Timestamp => timestamp.toInstant.toEpochMilli + // Binary is considered to be illegal here. + case _ => + throw new IllegalArgumentException( + s"Unsupported type for parsing to Long: ${fieldValue.getClass.getName}" + ) + } + } + attempt.recover { + case e: Exception => throw new AttributeTypeException( - s"not able to parse type ${fieldValue.getClass} to Long: ${fieldValue.toString}" + s"Failed to parse type ${fieldValue.getClass.getName} to Long: ${fieldValue.toString}", + e ) - } + }.get } @throws[AttributeTypeException] def parseTimestamp(fieldValue: Any): Timestamp = { - val parseError = new AttributeTypeException( - s"not able to parse type ${fieldValue.getClass} to Timestamp: ${fieldValue.toString}" - ) - fieldValue match { - case str: String => new Timestamp(DateParserUtils.parseDate(str.trim).getTime) - case long: java.lang.Long => new Timestamp(long) - case timestamp: Timestamp => timestamp - case date: java.util.Date => new Timestamp(date.getTime) - // Integer, Double, Boolean, Binary are considered to be illegal here. - case _ => - throw parseError + val attempt: Try[Timestamp] = Try { + fieldValue match { + case str: String => new Timestamp(DateParserUtils.parseDate(str.trim).getTime) + case long: java.lang.Long => new Timestamp(long) + case timestamp: Timestamp => timestamp + case date: java.util.Date => new Timestamp(date.getTime) + // Integer, Double, Boolean, Binary are considered to be illegal here. + case _ => + throw new AttributeTypeException( + s"Unsupported type for parsing to Timestamp: ${fieldValue.getClass.getName}" + ) + } } + + attempt.recover { + case e: Exception => + throw new AttributeTypeException( + s"Failed to parse type ${fieldValue.getClass.getName} to Timestamp: ${fieldValue.toString}", + e + ) + }.get + } @throws[AttributeTypeException] def parseDouble(fieldValue: Any): java.lang.Double = { - fieldValue match { - case str: String => str.trim.toDouble - case int: Integer => int.toDouble - case long: java.lang.Long => long.toDouble - case double: java.lang.Double => double - case boolean: java.lang.Boolean => if (boolean) 1 else 0 - // Timestamp and Binary are considered to be illegal here. - case _ => + val attempt: Try[Double] = Try { + fieldValue match { + case str: String => str.trim.toDouble + case int: Integer => int.toDouble + case long: java.lang.Long => long.toDouble + case double: java.lang.Double => double + case boolean: java.lang.Boolean => if (boolean) 1 else 0 + // Timestamp and Binary are considered to be illegal here. + case _ => + throw new AttributeTypeException( + s"Unsupported type for parsing to Double: ${fieldValue.getClass.getName}" + ) + } + } + + attempt.recover { + case e: Exception => throw new AttributeTypeException( - s"not able to parse type ${fieldValue.getClass} to Double: ${fieldValue.toString}" + s"Failed to parse type ${fieldValue.getClass.getName} to Double: ${fieldValue.toString}", + e ) - } + }.get + } @throws[AttributeTypeException] private def parseBoolean(fieldValue: Any): java.lang.Boolean = { - val parseError = new AttributeTypeException( - s"not able to parse type ${fieldValue.getClass} to Boolean: ${fieldValue.toString}" - ) - fieldValue match { - case str: String => - (Try(str.trim.toBoolean) orElse Try(str.trim.toInt == 1)) - .getOrElse(throw parseError) - case int: Integer => int != 0 - case long: java.lang.Long => long != 0 - case double: java.lang.Double => double != 0 - case boolean: java.lang.Boolean => boolean - // Timestamp and Binary are considered to be illegal here. - case _ => - throw parseError + val attempt: Try[Boolean] = Try { + fieldValue match { + case str: String => + (Try(str.trim.toBoolean) orElse Try(str.trim.toInt == 1)).get + case int: Integer => int != 0 + case long: java.lang.Long => long != 0 + case double: java.lang.Double => double != 0 + case boolean: java.lang.Boolean => boolean + // Timestamp and Binary are considered to be illegal here. + case _ => + throw new AttributeTypeException( + s"Unsupported type for parsing to Boolean: ${fieldValue.getClass.getName}" + ) + } } + + attempt.recover { + case e: Exception => + throw new AttributeTypeException( + s"Failed to parse type ${fieldValue.getClass.getName} to Boolean: ${fieldValue.toString}", + e + ) + }.get } /** @@ -203,9 +268,9 @@ object AttributeTypeUtils extends Serializable { * @return */ private def inferRow( - attributeTypes: Array[AttributeType], - fields: Array[Any] - ): Unit = { + attributeTypes: Array[AttributeType], + fields: Array[Any] + ): Unit = { for (i <- fields.indices) { attributeTypes.update(i, inferField(attributeTypes.apply(i), fields.apply(i))) } @@ -306,5 +371,6 @@ object AttributeTypeUtils extends Serializable { } } - class AttributeTypeException(msg: String) extends IllegalArgumentException(msg) {} -} + class AttributeTypeException(msg: String, cause: Throwable = null) + extends IllegalArgumentException(msg, cause) {} +} \ No newline at end of file diff --git a/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala b/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala index 62607eb4ccb..bd7450cc12e 100644 --- a/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala +++ b/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala @@ -1,7 +1,12 @@ package edu.uci.ics.amber.core.tuple -import edu.uci.ics.amber.core.tuple.AttributeTypeUtils.{inferField, inferSchemaFromRows} import edu.uci.ics.amber.core.tuple.AttributeType._ +import edu.uci.ics.amber.core.tuple.AttributeTypeUtils.{ + AttributeTypeException, + inferField, + inferSchemaFromRows, + parseField +} import org.scalatest.funsuite.AnyFunSuite class AttributeTypeUtilsSpec extends AnyFunSuite { @@ -83,4 +88,87 @@ class AttributeTypeUtilsSpec extends AnyFunSuite { } -} + test("parseField correctly parses to INTEGER") { + assert(parseField("123", AttributeType.INTEGER) == 123) + assert(parseField("1,234", AttributeType.INTEGER, force = true) == 1234) + assert(parseField(456, AttributeType.INTEGER) == 456) + assert(parseField(123.45, AttributeType.INTEGER) == 123) + assert(parseField(true, AttributeType.INTEGER) == 1) + assert(parseField(false, AttributeType.INTEGER) == 0) + assertThrows[AttributeTypeException] { + parseField("invalid", AttributeType.INTEGER) + } + assertThrows[AttributeTypeException] { + parseField("1,234", AttributeType.INTEGER) + } + } + + test("parseField correctly parses to LONG") { + assert(parseField("1234567890", AttributeType.LONG) == 1234567890L) + assert(parseField("1,234,567", AttributeType.LONG, force = true) == 1234567L) + assert(parseField(12345L, AttributeType.LONG) == 12345L) + assert(parseField(123.45, AttributeType.LONG) == 123L) + assert(parseField(true, AttributeType.LONG) == 1L) + assertThrows[AttributeTypeException] { + parseField("invalid", AttributeType.LONG) + } + assertThrows[AttributeTypeException] { + parseField("1,234,567", AttributeType.LONG) + } + } + + test("parseField correctly parses to DOUBLE") { + assert(parseField("123.45", AttributeType.DOUBLE) == 123.45) + assert(parseField(12345, AttributeType.DOUBLE) == 12345.0) + assert(parseField(12345L, AttributeType.DOUBLE) == 12345.0) + assert(parseField(true, AttributeType.DOUBLE) == 1.0) + assertThrows[AttributeTypeException] { + parseField("invalid", AttributeType.DOUBLE) + } + } + + test("parseField correctly parses to BOOLEAN") { + assert(parseField("true", AttributeType.BOOLEAN) == true) + assert(parseField("True", AttributeType.BOOLEAN) == true) + assert(parseField("TRUE", AttributeType.BOOLEAN) == true) + assert(parseField("false", AttributeType.BOOLEAN) == false) + assert(parseField("False", AttributeType.BOOLEAN) == false) + assert(parseField("FALSE", AttributeType.BOOLEAN) == false) + assert(parseField("1", AttributeType.BOOLEAN) == true) + assert(parseField("0", AttributeType.BOOLEAN) == false) + assert(parseField(1, AttributeType.BOOLEAN) == true) + assert(parseField(0, AttributeType.BOOLEAN) == false) + assertThrows[AttributeTypeException] { + parseField("invalid", AttributeType.BOOLEAN) + } + } + + test("parseField correctly parses to TIMESTAMP") { + val timestamp = + parseField("2023-11-13T10:15:30", AttributeType.TIMESTAMP).asInstanceOf[java.sql.Timestamp] + assert(timestamp.toString == "2023-11-13 10:15:30.0") + + assert( + parseField(1699820130000L, AttributeType.TIMESTAMP) + .asInstanceOf[java.sql.Timestamp] + .getTime == 1699820130000L + ) + + assertThrows[AttributeTypeException] { + parseField("invalid", AttributeType.TIMESTAMP) + } + } + + test("parseField correctly parses to STRING") { + assert(parseField(123, AttributeType.STRING) == "123") + assert(parseField(123.45, AttributeType.STRING) == "123.45") + assert(parseField(true, AttributeType.STRING) == "true") + } + + test("parseField returns original value for BINARY and ANY") { + val binaryData = Array[Byte](1, 2, 3) + assert(parseField(binaryData, AttributeType.BINARY) == binaryData) + assert(parseField("anything", AttributeType.ANY) == "anything") + } + +} \ No newline at end of file From fb3fff2cd8c026cecbbc5fbebb4ea272d74cafe6 Mon Sep 17 00:00:00 2001 From: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com> Date: Thu, 28 Nov 2024 22:58:11 -0500 Subject: [PATCH 2/2] fix format --- .../amber/core/tuple/AttributeTypeUtils.scala | 42 +++++++++---------- .../core/tuple/AttributeTypeUtilsSpec.scala | 2 +- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala index 12f21cb6597..1e333b8d7de 100644 --- a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala +++ b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtils.scala @@ -21,10 +21,10 @@ object AttributeTypeUtils extends Serializable { * @return schema of data */ def SchemaCasting( - schema: Schema, - attribute: String, - resultType: AttributeType - ): Schema = { + schema: Schema, + attribute: String, + resultType: AttributeType + ): Schema = { // need a builder to maintain the order of original schema val builder = Schema.builder() val attributes: List[Attribute] = schema.getAttributes @@ -33,8 +33,8 @@ object AttributeTypeUtils extends Serializable { if (attributes.apply(i).getName.equals(attribute)) { resultType match { case AttributeType.STRING | AttributeType.INTEGER | AttributeType.DOUBLE | - AttributeType.LONG | AttributeType.BOOLEAN | AttributeType.TIMESTAMP | - AttributeType.BINARY => + AttributeType.LONG | AttributeType.BOOLEAN | AttributeType.TIMESTAMP | + AttributeType.BINARY => builder.add(attribute, resultType) case AttributeType.ANY | _ => builder.add(attribute, attributes.apply(i).getType) @@ -60,9 +60,9 @@ object AttributeTypeUtils extends Serializable { * as specified by the typeCastingUnits. */ def tupleCasting( - tuple: Tuple, - targetTypes: Map[String, AttributeType] - ): TupleLike = + tuple: Tuple, + targetTypes: Map[String, AttributeType] + ): TupleLike = TupleLike( tuple.getSchema.getAttributes.map { attr => val targetType = targetTypes.getOrElse(attr.getName, attr.getType) @@ -82,9 +82,9 @@ object AttributeTypeUtils extends Serializable { */ @throws[AttributeTypeException] def parseFields( - fields: Array[Any], - attributeTypes: Array[AttributeType] - ): Array[Any] = { + fields: Array[Any], + attributeTypes: Array[AttributeType] + ): Array[Any] = { fields.indices.map(i => parseField(fields(i), attributeTypes(i))).toArray } @@ -99,10 +99,10 @@ object AttributeTypeUtils extends Serializable { */ @throws[AttributeTypeException] def parseField( - field: Any, - attributeType: AttributeType, - force: Boolean = false - ): Any = { + field: Any, + attributeType: AttributeType, + force: Boolean = false + ): Any = { if (field == null) return null attributeType match { case AttributeType.INTEGER => parseInteger(field, force) @@ -268,9 +268,9 @@ object AttributeTypeUtils extends Serializable { * @return */ private def inferRow( - attributeTypes: Array[AttributeType], - fields: Array[Any] - ): Unit = { + attributeTypes: Array[AttributeType], + fields: Array[Any] + ): Unit = { for (i <- fields.indices) { attributeTypes.update(i, inferField(attributeTypes.apply(i), fields.apply(i))) } @@ -372,5 +372,5 @@ object AttributeTypeUtils extends Serializable { } class AttributeTypeException(msg: String, cause: Throwable = null) - extends IllegalArgumentException(msg, cause) {} -} \ No newline at end of file + extends IllegalArgumentException(msg, cause) {} +} diff --git a/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala b/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala index bd7450cc12e..1c9bd49639f 100644 --- a/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala +++ b/core/workflow-core/src/test/scala/edu/uci/ics/amber/core/tuple/AttributeTypeUtilsSpec.scala @@ -171,4 +171,4 @@ class AttributeTypeUtilsSpec extends AnyFunSuite { assert(parseField("anything", AttributeType.ANY) == "anything") } -} \ No newline at end of file +}