<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.sql.hive

import java.rmi.server.UID
import java.util.{Properties, ArrayList => JArrayList}

import scala.collection.JavaConversions._
import scala.language.implicitConversions

import com.esotericsoftware.kryo.Kryo
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hive.common.StatsSetupConst
import org.apache.hadoop.hive.common.`type`.HiveDecimal
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.ql.Context
import org.apache.hadoop.hive.ql.exec.{UDF, Utilities}
import org.apache.hadoop.hive.ql.metadata.{Hive, Partition, Table}
import org.apache.hadoop.hive.ql.plan.{CreateTableDesc, FileSinkDesc, TableDesc}
import org.apache.hadoop.hive.ql.processors.CommandProcessorFactory
import org.apache.hadoop.hive.serde.serdeConstants
import org.apache.hadoop.hive.serde2.avro.AvroGenericRecordWritable
import org.apache.hadoop.hive.serde2.objectinspector.primitive.{HiveDecimalObjectInspector, PrimitiveObjectInspectorFactory}
import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorConverters, PrimitiveObjectInspector}
import org.apache.hadoop.hive.serde2.typeinfo.{DecimalTypeInfo, TypeInfo, TypeInfoFactory}
import org.apache.hadoop.hive.serde2.{ColumnProjectionUtils, Deserializer, io => hiveIo}
import org.apache.hadoop.io.{NullWritable, Writable}
import org.apache.hadoop.mapred.InputFormat
import org.apache.hadoop.{io => hadoopIo}

import org.apache.spark.Logging
import org.apache.spark.sql.types.{Decimal, DecimalType, UTF8String}

/**
 * This class provides the UDF creation and also the UDF instance serialization and
 * de-serialization cross process boundary.
 * 
 * Detail discussion can be found at https://github.com/apache/spark/pull/3640
 *
 * @param functionClassName UDF class name
 */
private[hive] case class HiveFunctionWrapper(var functionClassName: String)
  extends java.io.Externalizable {

  // for Serialization
  def this() = this(null)

  import org.apache.spark.util.Utils._

  @transient
  private val methodDeSerialize = {
    val method = classOf[Utilities].getDeclaredMethod(
      "deserializeObjectByKryo",
      classOf[Kryo],
      classOf[java.io.InputStream],
      classOf[Class[_]])
    method.setAccessible(true)

    method
  }

  @transient
  private val methodSerialize = {
    val method = classOf[Utilities].getDeclaredMethod(
      "serializeObjectByKryo",
      classOf[Kryo],
      classOf[Object],
      classOf[java.io.OutputStream])
    method.setAccessible(true)

    method
  }

  def deserializePlan[UDFType](is: java.io.InputStream, clazz: Class[_]): UDFType = {
    methodDeSerialize.invoke(null, Utilities.runtimeSerializationKryo.get(), is, clazz)
      .asInstanceOf[UDFType]
  }

  def serializePlan(function: AnyRef, out: java.io.OutputStream): Unit = {
    methodSerialize.invoke(null, Utilities.runtimeSerializationKryo.get(), function, out)
  }

  private var instance: AnyRef = null

  def writeExternal(out: java.io.ObjectOutput) {
    // output the function name
    out.writeUTF(functionClassName)

    // Write a flag if instance is null or not
    out.writeBoolean(instance != null)
    if (instance != null) {
      // Some of the UDF are serializable, but some others are not<commit_msg>[SPARK-6505] [SQL] Remove the reflection call in HiveFunctionWrapper<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.sql.hive

import java.rmi.server.UID
import java.util.{Properties, ArrayList => JArrayList}
import java.io.{OutputStream, InputStream}

import scala.collection.JavaConversions._
import scala.language.implicitConversions
import scala.reflect.ClassTag

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.Input
import com.esotericsoftware.kryo.io.Output
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hive.common.StatsSetupConst
import org.apache.hadoop.hive.common.`type`.HiveDecimal
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.ql.Context
import org.apache.hadoop.hive.ql.exec.{UDF, Utilities}
import org.apache.hadoop.hive.ql.metadata.{Hive, Partition, Table}
import org.apache.hadoop.hive.ql.plan.{CreateTableDesc, FileSinkDesc, TableDesc}
import org.apache.hadoop.hive.ql.processors.CommandProcessorFactory
import org.apache.hadoop.hive.serde.serdeConstants
import org.apache.hadoop.hive.serde2.avro.AvroGenericRecordWritable
import org.apache.hadoop.hive.serde2.objectinspector.primitive.{HiveDecimalObjectInspector, PrimitiveObjectInspectorFactory}
import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorConverters, PrimitiveObjectInspector}
import org.apache.hadoop.hive.serde2.typeinfo.{DecimalTypeInfo, TypeInfo, TypeInfoFactory}
import org.apache.hadoop.hive.serde2.{ColumnProjectionUtils, Deserializer, io => hiveIo}
import org.apache.hadoop.io.{NullWritable, Writable}
import org.apache.hadoop.mapred.InputFormat
import org.apache.hadoop.{io => hadoopIo}

import org.apache.spark.Logging
import org.apache.spark.sql.types.{Decimal, DecimalType, UTF8String}
import org.apache.spark.util.Utils._

/**
 * This class provides the UDF creation and also the UDF instance serialization and
 * de-serialization cross process boundary.
 * 
 * Detail discussion can be found at https://github.com/apache/spark/pull/3640
 *
 * @param functionClassName UDF class name
 */
private[hive] case class HiveFunctionWrapper(var functionClassName: String)
  extends java.io.Externalizable {

  // for Serialization
  def this() = this(null)

  @transient
  def deserializeObjectByKryo[T: ClassTag](
      kryo: Kryo,
      in: InputStream,
      clazz: Class[_]): T = {
    val inp = new Input(in)
    val t: T = kryo.readObject(inp,clazz).asInstanceOf[T]
    inp.close()
    t
  }

  @transient
  def serializeObjectByKryo(
      kryo: Kryo,
      plan: Object,
      out: OutputStream ) {
    val output: Output = new Output(out)
    kryo.writeObject(output, plan)
    output.close()
  }

  def deserializePlan[UDFType](is: java.io.InputStream, clazz: Class[_]): UDFType = {
    deserializeObjectByKryo(Utilities.runtimeSerializationKryo.get(), is, clazz)
      .asInstanceOf[UDFType]
  }

  def serializePlan(function: AnyRef, out: java.io.OutputStream): Unit = {
    serializeObjectByKryo(Utilities.runtimeSerializationKryo.get(), function, out)
  }

  private var instance: AnyRef = null

  def writeExternal(out: java.io.ObjectOutput) {
    // output the function name
    out.writeUTF(functionClassName)

    // Write a flag if instance is null or not
    out.writeBoolean(instance != null)
    if (instance != null) {
      // Some of the UDF are serializable, but some others are not
{'fraction_non_alphanumeric': 0.09578038839606809, 'fraction_numerical': 0.0038360105490290097, 'mean_word_length': 4.52882703777336, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 2, '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': '2277889', 'n_tokens_mistral': 2525, 'n_tokens_neox': 2504, 'n_words': 655}