<commit_before>/*
 * Copyright 2017 Datamountaineer.
 *
 * Licensed 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 com.datamountaineer.streamreactor.connect.mongodb.converters

import java.nio.ByteBuffer
import java.text.SimpleDateFormat
import java.util
import java.util.TimeZone

import org.apache.kafka.connect.data._
import org.apache.kafka.connect.errors.DataException
import org.apache.kafka.connect.sink.SinkRecord
import org.bson.Document
import org.json4s.JValue
import org.json4s.JsonAST._

import scala.collection.JavaConversions._

object SinkRecordConverter {
  private val ISO_DATE_FORMAT: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
  private val TIME_FORMAT: SimpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSSZ")

  ISO_DATE_FORMAT.setTimeZone(TimeZone.getTimeZone("UTC"))

  /**
    * Creates a Mongo document from a HashMap
    *
    * @param map
    * @return
    */
  def fromMap(map: util.HashMap[String, AnyRef]): Document = new Document(map)

  /**
    * Creates a Mongo document from a the Kafka Struct
    *
    * @param record
    * @return
    */
  def fromStruct(record: SinkRecord): Document = {

    def convertToDocument(schema: Schema, value: Any): Any = {
      Option(value) match {
        case None =>
          if (schema == null) null
          else {
            if (schema.defaultValue != null) convertToDocument(schema, schema.defaultValue)
            else if (schema.isOptional) null
            else throw new DataException("Conversion error: null value for field that is required and has no default value")
          }

        case Some(_) =>

          try {
            val schemaType = Option(schema).map(_.`type`())
              .orElse(Option(ConnectSchema.schemaType(value.getClass)))
              .getOrElse(throw new DataException("Class " + value.getClass + " does not have corresponding schema type."))

            schemaType match {

              case Schema.Type.INT8 | Schema.Type.INT16 | Schema.Type.BOOLEAN | Schema.Type.FLOAT32 | Schema.Type.FLOAT64 => value
              case Schema.Type.INT32 =>
                if (schema != null && Date.LOGICAL_NAME == schema.name) ISO_DATE_FORMAT.format(Date.toLogical(schema, value.asInstanceOf[Int]))
                else if (schema != null && Time.LOGICAL_NAME == schema.name) TIME_FORMAT.format(Time.toLogical(schema, value.asInstanceOf[Int]))
                else value

              case Schema.Type.INT64 =>
                if (schema != null && Timestamp.LOGICAL_NAME == schema.name) Timestamp.toLogical(schema, value.asInstanceOf[Long])
                else value

              case Schema.Type.STRING => value.asInstanceOf[CharSequence].toString

              case Schema.Type.BYTES =>
                if (schema != null && Decimal.LOGICAL_NAME == schema.name) Decimal.toLogical(schema, value.asInstanceOf[Array[Byte]])
                else value match {
                  case arrayByte: Array[Byte] => arrayByte
                  case buffer: ByteBuffer => buffer.array
                  case _ => throw new DataException("Invalid type for bytes type: " + value.getClass)
                }

              case Schema.Type.ARRAY =>
                val valueSchema = Option(schema).map(_.valueSchema()).orNull
                val list = new java.util.ArrayList[Any]
                value
                  .asInstanceOf[java.util.Collection[_]]
                  .foreach { elem =>
                    list.add(convertToDocument(valueSchema, elem))
                  }
                list

              case Schema.Type.MAP =>
                val map = value.asInstanceOf[java.util.Map[_, _]]
                // If true, using string keys and JSON object; if false, using non-string keys and Array-encoding
                val objectMode: Boolean = Option(schema)
                  .map(_.keySchema.`type` eq Schema.Type.STRING)
                  .getOrElse(map.entrySet().headOption.forall(_.getKey.isInstanceOf[String]))

                var obj: Document = null
                var list: java.util.ArrayList[Any] = null

                if (objectMode) obj = new Document()
                else list = new util.ArrayList[Any]()

                for (entry <- map.entrySet) {
                  val keySchema = Option(schema).map(_.keySchema()).orNull
                  val valueSchema = Option(schema).map(_.valueSchema).orNull

                  val mapKey = convertToDocument(keySchema, entry.getKey)
                  //don't add null
                  Option(convertToDocument(valueSchema, entry.getValue))
                    .foreach { mapValue =>
                      if (objectMode) {
                        obj.append(mapKey.toString, mapValue)
                      }
                      else {
                        val innerArray = new util.ArrayList[Any]()
                        innerArray.add(mapKey)
                        innerArray.add(mapValue)
                        list.add(innerArray)
                      }
                    }
                }
                if (objectMode) obj
                else list

              case Schema.Type.STRUCT =>
                val struct = value.asInstanceOf[Struct]
                if (struct.schema != schema) throw new DataException("Mismatching schema.")

                schema.fields
                  .foldLeft(new Document) { (document, field) =>
                    Option(convertToDocument(field.schema, struct.get(field)))
                      .foreach(//case bd: BigDecimal => document.append(field.name, bd.toDouble)
                        //case bi: BigInt => document.append(field.name(), bi.toLong)
                        v => document.append(field.name, v))
                    document
                  }

              case other =>
                schema.name() match {
                  case Decimal.LOGICAL_NAME =>
                    value match {
                      case bd: BigDecimal => bd
                      case bb: ByteBuffer => Decimal.toLogical(schema, bb.array())
                      case _ => Decimal.toLogical(schema, value.asInstanceOf[Array[Byte]])
                    }
                  case Date.LOGICAL_NAME =>
                    value match {
                      case d: Date => ISO_DATE_FORMAT.format(d)
                      case _ => ISO_DATE_FORMAT.format(Date.toLogical(schema, value.asInstanceOf[Int]))
                    }

                  case Time.LOGICAL_NAME =>
                    value match {
                      case d: Date => TIME_FORMAT.format(d)
                      case _ => TIME_FORMAT.format(Time.toLogical(schema, value.asInstanceOf[Int]))
                    }

                  case Timestamp.LOGICAL_NAME =>
                    value match {
                      case d: Date => ISO_DATE_FORMAT.format(d)
                      case _ => ISO_DATE_FORMAT.format(Timestamp.toLogical(schema, value.asInstanceOf[Long]))
                    }

                  case _ => sys.error(s"$other is not a recognized schema")
                }
            }
          }
          catch {
            case _: ClassCastException => throw new DataException("Invalid type for " + schema.`type` + ": " + value.getClass)<commit_msg>fixing match type to use java.util.date<commit_after>/*
 * Copyright 2017 Datamountaineer.
 *
 * Licensed 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 com.datamountaineer.streamreactor.connect.mongodb.converters

import java.nio.ByteBuffer
import java.text.SimpleDateFormat
import java.util
import java.util.TimeZone

import org.apache.kafka.connect.data._
import org.apache.kafka.connect.errors.DataException
import org.apache.kafka.connect.sink.SinkRecord
import org.bson.Document
import org.json4s.JValue
import org.json4s.JsonAST._

import scala.collection.JavaConversions._

object SinkRecordConverter {
  private val ISO_DATE_FORMAT: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
  private val TIME_FORMAT: SimpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSSZ")

  ISO_DATE_FORMAT.setTimeZone(TimeZone.getTimeZone("UTC"))

  /**
    * Creates a Mongo document from a HashMap
    *
    * @param map
    * @return
    */
  def fromMap(map: util.HashMap[String, AnyRef]): Document = new Document(map)

  /**
    * Creates a Mongo document from a the Kafka Struct
    *
    * @param record
    * @return
    */
  def fromStruct(record: SinkRecord): Document = {

    def convertToDocument(schema: Schema, value: Any): Any = {
      Option(value) match {
        case None =>
          if (schema == null) null
          else {
            if (schema.defaultValue != null) convertToDocument(schema, schema.defaultValue)
            else if (schema.isOptional) null
            else throw new DataException("Conversion error: null value for field that is required and has no default value")
          }

        case Some(_) =>

          try {
            val schemaType = Option(schema).map(_.`type`())
              .orElse(Option(ConnectSchema.schemaType(value.getClass)))
              .getOrElse(throw new DataException("Class " + value.getClass + " does not have corresponding schema type."))

            schemaType match {

              case Schema.Type.INT8 | Schema.Type.INT16 | Schema.Type.BOOLEAN | Schema.Type.FLOAT32 | Schema.Type.FLOAT64 => value
              case Schema.Type.INT32 =>
                if (schema != null && Date.LOGICAL_NAME == schema.name) ISO_DATE_FORMAT.format(Date.toLogical(schema, value.asInstanceOf[Int]))
                else if (schema != null && Time.LOGICAL_NAME == schema.name) TIME_FORMAT.format(Time.toLogical(schema, value.asInstanceOf[Int]))
                else value

              case Schema.Type.INT64 =>
                if (schema != null && Timestamp.LOGICAL_NAME == schema.name) Timestamp.toLogical(schema, value.asInstanceOf[Long])
                else value

              case Schema.Type.STRING => value.asInstanceOf[CharSequence].toString

              case Schema.Type.BYTES =>
                if (schema != null && Decimal.LOGICAL_NAME == schema.name) Decimal.toLogical(schema, value.asInstanceOf[Array[Byte]])
                else value match {
                  case arrayByte: Array[Byte] => arrayByte
                  case buffer: ByteBuffer => buffer.array
                  case _ => throw new DataException("Invalid type for bytes type: " + value.getClass)
                }

              case Schema.Type.ARRAY =>
                val valueSchema = Option(schema).map(_.valueSchema()).orNull
                val list = new java.util.ArrayList[Any]
                value
                  .asInstanceOf[java.util.Collection[_]]
                  .foreach { elem =>
                    list.add(convertToDocument(valueSchema, elem))
                  }
                list

              case Schema.Type.MAP =>
                val map = value.asInstanceOf[java.util.Map[_, _]]
                // If true, using string keys and JSON object; if false, using non-string keys and Array-encoding
                val objectMode: Boolean = Option(schema)
                  .map(_.keySchema.`type` eq Schema.Type.STRING)
                  .getOrElse(map.entrySet().headOption.forall(_.getKey.isInstanceOf[String]))

                var obj: Document = null
                var list: java.util.ArrayList[Any] = null

                if (objectMode) obj = new Document()
                else list = new util.ArrayList[Any]()

                for (entry <- map.entrySet) {
                  val keySchema = Option(schema).map(_.keySchema()).orNull
                  val valueSchema = Option(schema).map(_.valueSchema).orNull

                  val mapKey = convertToDocument(keySchema, entry.getKey)
                  //don't add null
                  Option(convertToDocument(valueSchema, entry.getValue))
                    .foreach { mapValue =>
                      if (objectMode) {
                        obj.append(mapKey.toString, mapValue)
                      }
                      else {
                        val innerArray = new util.ArrayList[Any]()
                        innerArray.add(mapKey)
                        innerArray.add(mapValue)
                        list.add(innerArray)
                      }
                    }
                }
                if (objectMode) obj
                else list

              case Schema.Type.STRUCT =>
                val struct = value.asInstanceOf[Struct]
                if (struct.schema != schema) throw new DataException("Mismatching schema.")

                schema.fields
                  .foldLeft(new Document) { (document, field) =>
                    Option(convertToDocument(field.schema, struct.get(field)))
                      .foreach(//case bd: BigDecimal => document.append(field.name, bd.toDouble)
                        //case bi: BigInt => document.append(field.name(), bi.toLong)
                        v => document.append(field.name, v))
                    document
                  }

              case other =>
                schema.name() match {
                  case Decimal.LOGICAL_NAME =>
                    value match {
                      case bd: BigDecimal => bd
                      case bb: ByteBuffer => Decimal.toLogical(schema, bb.array())
                      case _ => Decimal.toLogical(schema, value.asInstanceOf[Array[Byte]])
                    }
                  case Date.LOGICAL_NAME =>
                    value match {
                      case d: java.util.Date => ISO_DATE_FORMAT.format(d)
                      case _ => ISO_DATE_FORMAT.format(Date.toLogical(schema, value.asInstanceOf[Int]))
                    }

                  case Time.LOGICAL_NAME =>
                    value match {
                      case d: java.util.Date => TIME_FORMAT.format(d)
                      case _ => TIME_FORMAT.format(Time.toLogical(schema, value.asInstanceOf[Int]))
                    }

                  case Timestamp.LOGICAL_NAME =>
                    value match {
                      case d: java.util.Date => ISO_DATE_FORMAT.format(d)
                      case _ => ISO_DATE_FORMAT.format(Timestamp.toLogical(schema, value.asInstanceOf[Long]))
                    }

                  case _ => sys.error(s"$other is not a recognized schema")
                }
            }
          }
          catch {
            case _: ClassCastException => throw new DataException("Invalid type for " + schema.`type` + ": " + value.getClass)
{'fraction_non_alphanumeric': 0.09593098120010302, 'fraction_numerical': 0.002704094772083441, 'mean_word_length': 1.7501770538243626, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3234946', 'n_tokens_mistral': 4095, 'n_tokens_neox': 3859, 'n_words': 972}