<commit_before>package shark.execution.serialization

import java.beans.{XMLDecoder, XMLEncoder, PersistenceDelegate}
import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectOutput, ObjectInput}

import com.ning.compress.lzf.{LZFEncoder, LZFDecoder}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hive.ql.exec.Utilities.EnumDelegate
import org.apache.hadoop.hive.ql.plan.GroupByDesc
import org.apache.hadoop.hive.ql.plan.PlanUtils.ExpressionTypes

import shark.SharkConfVars


/**
 * Java object serialization using XML encoder/decoder. Avoid using this to
 * serialize byte arrays because it is extremely inefficient.
 */
object XmlSerializer {
  // We prepend the buffer with a byte indicating whether payload is compressed
  val COMPRESSION_ENABLED : Byte = 1;
  val COMPRESSION_DISABLED : Byte = 0;

  def serialize[T](o: T, conf: Configuration): Array[Byte] = {
    val byteStream = new ByteArrayOutputStream()
    val e = new XMLEncoder(byteStream)
    // workaround for java 1.5
    e.setPersistenceDelegate(classOf[ExpressionTypes], new EnumDelegate())
    e.setPersistenceDelegate(classOf[GroupByDesc.Mode], new EnumDelegate())
    e.writeObject(o)
    e.close()

    val useCompression = conf match {
      case null => SharkConfVars.COMPRESS_QUERY_PLAN.defaultBoolVal
      case _ => SharkConfVars.getBoolVar(conf, SharkConfVars.COMPRESS_QUERY_PLAN)
    }

    if (useCompression) {
      COMPRESSION_ENABLED +: LZFEncoder.encode(byteStream.toByteArray())
    } else {
      COMPRESSION_DISABLED +: byteStream.toByteArray
    }
  }

  def deserialize[T](bytes: Array[Byte]): T  = {
    val cl = Thread.currentThread.getContextClassLoader
    val decodedStream =
      if (bytes(0) == COMPRESSION_ENABLED) {
        new ByteArrayInputStream(LZFDecoder.decode(bytes.slice(1, bytes.size)))
      } else {
        new ByteArrayInputStream(bytes.slice(1, bytes.size))
      }
    val d: XMLDecoder = new XMLDecoder(decodedStream, null, null, cl)
    val ret = d.readObject()
    d.close()
    ret.asInstanceOf[T]
  }
}<commit_msg>SHARK-95: xml deserialization should be synchronized.<commit_after>package shark.execution.serialization

import java.beans.{XMLDecoder, XMLEncoder, PersistenceDelegate}
import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectOutput, ObjectInput}

import com.ning.compress.lzf.{LZFEncoder, LZFDecoder}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hive.ql.exec.Utilities.EnumDelegate
import org.apache.hadoop.hive.ql.plan.GroupByDesc
import org.apache.hadoop.hive.ql.plan.PlanUtils.ExpressionTypes

import shark.{SharkConfVars, SharkEnvSlave}


/**
 * Java object serialization using XML encoder/decoder. Avoid using this to
 * serialize byte arrays because it is extremely inefficient.
 */
object XmlSerializer {
  // We prepend the buffer with a byte indicating whether payload is compressed
  val COMPRESSION_ENABLED: Byte = 1
  val COMPRESSION_DISABLED: Byte = 0

  def serialize[T](o: T, conf: Configuration): Array[Byte] = {
    val byteStream = new ByteArrayOutputStream()
    val e = new XMLEncoder(byteStream)
    // workaround for java 1.5
    e.setPersistenceDelegate(classOf[ExpressionTypes], new EnumDelegate())
    e.setPersistenceDelegate(classOf[GroupByDesc.Mode], new EnumDelegate())
    e.writeObject(o)
    e.close()

    val useCompression = conf match {
      case null => SharkConfVars.COMPRESS_QUERY_PLAN.defaultBoolVal
      case _ => SharkConfVars.getBoolVar(conf, SharkConfVars.COMPRESS_QUERY_PLAN)
    }

    if (useCompression) {
      COMPRESSION_ENABLED +: LZFEncoder.encode(byteStream.toByteArray())
    } else {
      COMPRESSION_DISABLED +: byteStream.toByteArray
    }
  }

  def deserialize[T](bytes: Array[Byte]): T  = {
    val cl = Thread.currentThread.getContextClassLoader
    val decodedStream =
      if (bytes(0) == COMPRESSION_ENABLED) {
        new ByteArrayInputStream(LZFDecoder.decode(bytes.slice(1, bytes.size)))
      } else {
        new ByteArrayInputStream(bytes.slice(1, bytes.size))
      }

    // Occasionally an object inspector is created from the decoding.
    // Need to put a lock on the process.
    val ret = SharkEnvSlave.objectInspectorLock.synchronized {
      val d: XMLDecoder = new XMLDecoder(decodedStream, null, null, cl)
      val ret = d.readObject()
      d.close()
      ret
    }
    ret.asInstanceOf[T]
  }
}
{'fraction_non_alphanumeric': 0.09678888635846049, 'fraction_numerical': 0.0036438168982008655, 'mean_word_length': 4.002277904328018, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, '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': '1218152', 'n_tokens_mistral': 1373, 'n_tokens_neox': 1360, 'n_words': 305}