<commit_before>
/*
 * Copyright 2015 ligaDATA
 *
 * 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.ligadata.KamanjaBase
import java.net.URL
import java.net.URLClassLoader
import java.io.{ ByteArrayInputStream, DataInputStream, DataOutputStream, ByteArrayOutputStream }
import com.ligadata.Exceptions.StackTrace
import org.apache.log4j._
import java.util.Date
import java.util.Calendar
import java.text.SimpleDateFormat

trait MessageContainerBase {
  // System Columns
  var transactionId: Long
  var timePartitionData: Long = 0 // By default we are taking Java date with 0 milliseconds.
  var rowNumber: Int = 0 // This is unique value with in transactionId

  // System Attributes Functions
  final def TransactionId(transId: Long): Unit = { transactionId = transId }
  final def TransactionId(): Long = transactionId

  final def TimePartitionData(timeInMillisecs: Long): Unit = { timePartitionData = timeInMillisecs }
  final def TimePartitionData(): Long = timePartitionData

  final def RowNumber(rno: Int): Unit = { rowNumber = rno }
  final def RowNumber(): Int = rowNumber

  def isMessage: Boolean
  def isContainer: Boolean
  def IsFixed: Boolean
  def IsKv: Boolean
  def CanPersist: Boolean
  def populate(inputdata: InputData): Unit
  def set(key: String, value: Any): Unit
  def get(key: String): Any
  def getOrElse(key: String, default: Any): Any
  def AddMessage(childPath: Array[(String, String)], msg: BaseMsg): Unit
  def GetMessage(childPath: Array[(String, String)], primaryKey: Array[String]): BaseMsg
  def Version: String // Message or Container Version
  def PartitionKeyData: Array[String] // Partition key data
  def PrimaryKeyData: Array[String] // Primary key data
  def FullName: String // Message or Container Full Name
  def NameSpace: String // Message or Container NameSpace
  def Name: String // Message or Container Name
  def Deserialize(dis: DataInputStream, mdResolver: MdBaseResolveInfo, loader: java.lang.ClassLoader, savedDataVersion: String): Unit
  def Serialize(dos: DataOutputStream): Unit
  def Save(): Unit
  def Clone(): MessageContainerBase
  def hasPrimaryKey: Boolean
  def hasPartitionKey: Boolean
  def hasTimeParitionInfo: Boolean
  def getNativeKeyValues: scala.collection.immutable.Map[String, (String, Any)]
}

trait MessageContainerObjBase {
  def isMessage: Boolean
  def isContainer: Boolean
  def IsFixed: Boolean
  def IsKv: Boolean
  def CanPersist: Boolean
  def FullName: String // Message or Container FullName
  def NameSpace: String // Message or Container NameSpace
  def Name: String // Message or Container Name
  def Version: String // Message or Container Version
  def PartitionKeyData(inputdata: InputData): Array[String] // Partition key data
  def PrimaryKeyData(inputdata: InputData): Array[String] // Primary key data
  def getTimePartitionInfo: (String, String, String) // FieldName, Format & Time Partition Types(Daily/Monthly/Yearly)
  def TimePartitionData(inputdata: InputData): Long
  def hasPrimaryKey: Boolean
  def hasPartitionKey: Boolean
  def hasTimeParitionInfo: Boolean

  private def extractTime(fieldData: String, timeFormat: String): Long = {
    if (fieldData == null || fieldData.trim() == "") return 0

    if (timeFormat == null || timeFormat.trim() == "") return 0

    if (timeFormat.compareToIgnoreCase("epochtimeInMillis") == 0)
      return fieldData.toLong

    if (timeFormat.compareToIgnoreCase("epochtimeInSeconds") == 0 || timeFormat.compareToIgnoreCase("epochtime") == 0)
      return fieldData.toLong * 1000

    // Now assuming Date partition format exists.
    val dtFormat = new SimpleDateFormat(timeFormat);
    val tm =
      if (fieldData.size == 0) {
        new Date(0)
      } else {
        dtFormat.parse(fieldData)
      }
    tm.getTime()
  }

  def ComputeTimePartitionData(fieldData: String, timeFormat: String, timePartitionType: String): Long = {
    val fldTimeDataInMs = extractTime(fieldData, timeFormat)

    // Align to Partition
    var cal: Calendar = Calendar.getInstance();
    cal.setTime(new Date(fldTimeDataInMs));

    if (timePartitionType == null || timePartitionType.trim() == "") return 0

    timePartitionType.toLowerCase match {
      case "yearly" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.set(Calendar.YEAR, cal.get(Calendar.YEAR));
        return newcal.getTime().getTime()
      }
      case "monthly" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.set(Calendar.YEAR, cal.get(Calendar.YEAR))
        newcal.set(Calendar.MONTH, cal.get(Calendar.MONTH))
        return newcal.getTime().getTime()
      }
      case "daily" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.set(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH), cal.get(Calendar.DAY_OF_MONTH))
        return newcal.getTime().getTime()
      }
    }
    return 0
  }

}

trait MdBaseResolveInfo {
  def getMessgeOrContainerInstance(typName: String): MessageContainerBase
}

object SerializeDeserialize {
  val loggerName = this.getClass.getName
  val logger = Logger.getLogger(loggerName)
  def Serialize(inst: MessageContainerBase): Array[Byte] = {
    val bos: ByteArrayOutputStream = new ByteArrayOutputStream(1024 * 1024)
    val dos = new DataOutputStream(bos)

    try {
      dos.writeUTF(inst.FullName)
      dos.writeUTF(inst.Version)
      dos.writeUTF(inst.getClass.getName)
      inst.Serialize(dos)
      val arr = bos.toByteArray
      dos.close
      bos.close
      return arr<commit_msg>TimePartition computation update - setting calender time to 0 before computing yearly, monthly and daily<commit_after>
/*
 * Copyright 2015 ligaDATA
 *
 * 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.ligadata.KamanjaBase
import java.net.URL
import java.net.URLClassLoader
import java.io.{ ByteArrayInputStream, DataInputStream, DataOutputStream, ByteArrayOutputStream }
import com.ligadata.Exceptions.StackTrace
import org.apache.log4j._
import java.util.Date
import java.util.Calendar
import java.text.SimpleDateFormat

trait MessageContainerBase {
  // System Columns
  var transactionId: Long
  var timePartitionData: Long = 0 // By default we are taking Java date with 0 milliseconds.
  var rowNumber: Int = 0 // This is unique value with in transactionId

  // System Attributes Functions
  final def TransactionId(transId: Long): Unit = { transactionId = transId }
  final def TransactionId(): Long = transactionId

  final def TimePartitionData(timeInMillisecs: Long): Unit = { timePartitionData = timeInMillisecs }
  final def TimePartitionData(): Long = timePartitionData

  final def RowNumber(rno: Int): Unit = { rowNumber = rno }
  final def RowNumber(): Int = rowNumber

  def isMessage: Boolean
  def isContainer: Boolean
  def IsFixed: Boolean
  def IsKv: Boolean
  def CanPersist: Boolean
  def populate(inputdata: InputData): Unit
  def set(key: String, value: Any): Unit
  def get(key: String): Any
  def getOrElse(key: String, default: Any): Any
  def AddMessage(childPath: Array[(String, String)], msg: BaseMsg): Unit
  def GetMessage(childPath: Array[(String, String)], primaryKey: Array[String]): BaseMsg
  def Version: String // Message or Container Version
  def PartitionKeyData: Array[String] // Partition key data
  def PrimaryKeyData: Array[String] // Primary key data
  def FullName: String // Message or Container Full Name
  def NameSpace: String // Message or Container NameSpace
  def Name: String // Message or Container Name
  def Deserialize(dis: DataInputStream, mdResolver: MdBaseResolveInfo, loader: java.lang.ClassLoader, savedDataVersion: String): Unit
  def Serialize(dos: DataOutputStream): Unit
  def Save(): Unit
  def Clone(): MessageContainerBase
  def hasPrimaryKey: Boolean
  def hasPartitionKey: Boolean
  def hasTimeParitionInfo: Boolean
  def getNativeKeyValues: scala.collection.immutable.Map[String, (String, Any)]
}

trait MessageContainerObjBase {
  def isMessage: Boolean
  def isContainer: Boolean
  def IsFixed: Boolean
  def IsKv: Boolean
  def CanPersist: Boolean
  def FullName: String // Message or Container FullName
  def NameSpace: String // Message or Container NameSpace
  def Name: String // Message or Container Name
  def Version: String // Message or Container Version
  def PartitionKeyData(inputdata: InputData): Array[String] // Partition key data
  def PrimaryKeyData(inputdata: InputData): Array[String] // Primary key data
  def getTimePartitionInfo: (String, String, String) // FieldName, Format & Time Partition Types(Daily/Monthly/Yearly)
  def TimePartitionData(inputdata: InputData): Long
  def hasPrimaryKey: Boolean
  def hasPartitionKey: Boolean
  def hasTimeParitionInfo: Boolean

  private def extractTime(fieldData: String, timeFormat: String): Long = {
    if (fieldData == null || fieldData.trim() == "") return 0

    if (timeFormat == null || timeFormat.trim() == "") return 0

    if (timeFormat.compareToIgnoreCase("epochtimeInMillis") == 0)
      return fieldData.toLong

    if (timeFormat.compareToIgnoreCase("epochtimeInSeconds") == 0 || timeFormat.compareToIgnoreCase("epochtime") == 0)
      return fieldData.toLong * 1000

    // Now assuming Date partition format exists.
    val dtFormat = new SimpleDateFormat(timeFormat);
    val tm =
      if (fieldData.size == 0) {
        new Date(0)
      } else {
        dtFormat.parse(fieldData)
      }
    tm.getTime()
  }

  def ComputeTimePartitionData(fieldData: String, timeFormat: String, timePartitionType: String): Long = {
    val fldTimeDataInMs = extractTime(fieldData, timeFormat)

    // Align to Partition
    var cal: Calendar = Calendar.getInstance();
    cal.setTime(new Date(fldTimeDataInMs));

    if (timePartitionType == null || timePartitionType.trim() == "") return 0

    timePartitionType.toLowerCase match {
      case "yearly" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.setTimeInMillis(0)
        newcal.set(Calendar.YEAR, cal.get(Calendar.YEAR));
        return newcal.getTime().getTime()
      }
      case "monthly" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.setTimeInMillis(0)
        newcal.set(Calendar.YEAR, cal.get(Calendar.YEAR))
        newcal.set(Calendar.MONTH, cal.get(Calendar.MONTH))
        return newcal.getTime().getTime()
      }
      case "daily" => {
        var newcal: Calendar = Calendar.getInstance();
        newcal.setTimeInMillis(0)
        newcal.set(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH), cal.get(Calendar.DAY_OF_MONTH))
        return newcal.getTime().getTime()
      }
    }
    return 0
  }

}

trait MdBaseResolveInfo {
  def getMessgeOrContainerInstance(typName: String): MessageContainerBase
}

object SerializeDeserialize {
  val loggerName = this.getClass.getName
  val logger = Logger.getLogger(loggerName)
  def Serialize(inst: MessageContainerBase): Array[Byte] = {
    val bos: ByteArrayOutputStream = new ByteArrayOutputStream(1024 * 1024)
    val dos = new DataOutputStream(bos)

    try {
      dos.writeUTF(inst.FullName)
      dos.writeUTF(inst.Version)
      dos.writeUTF(inst.getClass.getName)
      inst.Serialize(dos)
      val arr = bos.toByteArray
      dos.close
      bos.close
      return arr
{'fraction_non_alphanumeric': 0.09344650289724965, 'fraction_numerical': 0.005712886640006529, 'mean_word_length': 3.8820717131474103, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '6964944', 'n_tokens_mistral': 3528, 'n_tokens_neox': 3430, 'n_words': 1089}