<commit_before>package com.asakusafw.spark.compiler

import org.junit.runner.RunWith
import org.scalatest.FlatSpec
import org.scalatest.junit.JUnitRunner

import java.io.{ File, DataInput, DataOutput }
import java.net.URLClassLoader
import java.util.{ List => JList }

import scala.collection.JavaConversions._

import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{ NullWritable, Writable }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.SparkContext._

import com.asakusafw.bridge.stage.StageInfo
import com.asakusafw.lang.compiler.api.CompilerOptions
import com.asakusafw.lang.compiler.api.testing.MockJobflowProcessorContext
import com.asakusafw.lang.compiler.model.PropertyName
import com.asakusafw.lang.compiler.model.description.ClassDescription
import com.asakusafw.lang.compiler.model.graph._
import com.asakusafw.lang.compiler.model.info.ExternalInputInfo
import com.asakusafw.lang.compiler.model.testing.OperatorExtractor
import com.asakusafw.lang.compiler.planning._
import com.asakusafw.runtime.compatibility.JobCompatibility
import com.asakusafw.runtime.core.Result
import com.asakusafw.runtime.model.DataModel
import com.asakusafw.runtime.stage.StageConstants
import com.asakusafw.runtime.stage.input.TemporaryInputFormat
import com.asakusafw.runtime.stage.output.TemporaryOutputFormat
import com.asakusafw.runtime.value._
import com.asakusafw.spark.runtime._
import com.asakusafw.spark.tools.asm._
import com.asakusafw.vocabulary.flow.processor.PartialAggregation
import com.asakusafw.vocabulary.model.{ Key, Summarized }
import com.asakusafw.vocabulary.operator._

@RunWith(classOf[JUnitRunner])
class SparkClientCompilerSpecTest extends SparkClientCompilerSpec

class SparkClientCompilerSpec extends FlatSpec with LoadClassSugar {

  import SparkClientCompilerSpec._

  behavior of classOf[SparkClientCompiler].getSimpleName

  it should "compile Spark client from simple plan" in {
    val tmpDir = File.createTempFile("test-", null)
    tmpDir.delete
    val classpath = new File(tmpDir, "classes").getAbsoluteFile
    classpath.mkdirs()
    val path = new File(tmpDir, "tmp").getAbsolutePath

    spark { sc =>
      val hoges = sc.parallelize(0 until 100).map { i =>
        val hoge = new Hoge()
        hoge.id.modify(i)
        hoge
      }
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      job.setOutputKeyClass(classOf[NullWritable])
      job.setOutputValueClass(classOf[Hoge])
      job.setOutputFormatClass(classOf[TemporaryOutputFormat[Hoge]])
      TemporaryOutputFormat.setOutputPath(job, new Path(path, "extenal/input/hoge"))
      hoges.map((NullWritable.get, _)).saveAsNewAPIHadoopDataset(job.getConfiguration)
    }

    val inputOperator = ExternalInput
      .newInstance("hoge/part-*",
        new ExternalInputInfo.Basic(
          ClassDescription.of(classOf[Hoge]),
          "test",
          ClassDescription.of(classOf[Hoge]),
          ExternalInputInfo.DataSize.UNKNOWN))

    val outputOperator = ExternalOutput
      .newInstance("output", inputOperator.getOperatorPort)

    val graph = new OperatorGraph(Seq(inputOperator, outputOperator))

    val compiler = new SparkClientCompiler {

      override def preparePlan(graph: OperatorGraph): Plan = {
        val plan = super.preparePlan(graph)
        assert(plan.getElements.size === 2)
        plan
      }
    }

    val jpContext = new MockJobflowProcessorContext(
      new CompilerOptions("buildid", path, Map.empty[String, String]),
      Thread.currentThread.getContextClassLoader,
      classpath)
    val jobflow = new Jobflow("flowId", ClassDescription.of(classOf[SparkClientCompilerSpec]), graph)

    compiler.process(jpContext, jobflow)

    val cl = Thread.currentThread.getContextClassLoader
    try {
      val classloader = new URLClassLoader(Array(classpath.toURI.toURL), cl)
      Thread.currentThread.setContextClassLoader(classloader)
      val cls = Class.forName("com.asakusafw.generated.spark.flowId.SparkClient", true, classloader)
        .asSubclass(classOf[SparkClient])
      val instance = cls.newInstance

      val conf = new SparkConf()
      conf.setAppName("AsakusaSparkClient")
      conf.setMaster("local[*]")

      val stageInfo = new StageInfo(
        sys.props("user.name"), "batchId", "flowId", null, "executionId", Map.empty[String, String])
      conf.setHadoopConf(Props.StageInfo, stageInfo.serialize)

      instance.execute(conf)
    } finally {
      Thread.currentThread.setContextClassLoader(cl)
    }

    spark { sc =>
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      TemporaryInputFormat.setInputPaths(job, Seq(new Path(path, s"output/*/part-*")))
      val rdd = sc.newAPIHadoopRDD(
        job.getConfiguration,
        classOf[TemporaryInputFormat[Hoge]],
        classOf[NullWritable],
        classOf[Hoge])
      assert(rdd.map(_._2.id.get).collect === (0 until 100))
    }
  }

  it should "compile Spark client with Extract" in {
    val tmpDir = File.createTempFile("test-", null)
    tmpDir.delete
    val classpath = new File(tmpDir, "classes").getAbsoluteFile
    classpath.mkdirs()
    val path = new File(tmpDir, "tmp").getAbsolutePath

    spark { sc =>
      val hoges = sc.parallelize(0 until 100).map { i =>
        val hoge = new Hoge()
        hoge.id.modify(i)
        hoge
      }
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      job.setOutputKeyClass(classOf[NullWritable])
      job.setOutputValueClass(classOf[Hoge])
      job.setOutputFormatClass(classOf[TemporaryOutputFormat[Hoge]])
      TemporaryOutputFormat.setOutputPath(job, new Path(path, "extenal/input/hoge"))
      hoges.map((NullWritable.get, _)).saveAsNewAPIHadoopDataset(job.getConfiguration)
    }

    val inputOperator = ExternalInput
      .newInstance("hoge/part-*",
        new ExternalInputInfo.Basic(
          ClassDescription.of(classOf[Hoge]),
          "test",
          ClassDescription.of(classOf[Hoge]),
          ExternalInputInfo.DataSize.UNKNOWN))

    val extractOperator = OperatorExtractor
      .extract(classOf[Extract], classOf[Ops], "extract")
      .input("hoge", ClassDescription.of(classOf[Hoge]), inputOperator.getOperatorPort)
      .output("evenResult", ClassDescription.of(classOf[Hoge]))
      .output("oddResult", ClassDescription.of(classOf[Hoge]))
      .build()

    val evenOutputOperator = ExternalOutput
      .newInstance("even", extractOperator.findOutput("evenResult"))

    val oddOutputOperator = ExternalOutput
      .newInstance("odd", extractOperator.findOutput("oddResult"))

    val graph = new OperatorGraph(Seq(
      inputOperator,
      extractOperator,
      evenOutputOperator,
      oddOutputOperator))

    val compiler = new SparkClientCompiler {

      override def preparePlan(graph: OperatorGraph): Plan = {
        val plan = super.preparePlan(graph)
        assert(plan.getElements.size === 4)
        plan
      }
    }

    val jpContext = new MockJobflowProcessorContext(
      new CompilerOptions("buildid", path, Map.empty[String, String]),
      Thread.currentThread.getContextClassLoader,
      classpath)
    val jobflow = new Jobflow("flowId", ClassDescription.of(classOf[SparkClientCompilerSpec]), graph)

    compiler.process(jpContext, jobflow)

    val cl = Thread.currentThread.getContextClassLoader
    try {
      val classloader = new URLClassLoader(Array(classpath.toURI.toURL), cl)
      Thread.currentThread.setContextClassLoader(classloader)
      val cls = Class.forName("com.asakusafw.generated.spark.flowId.SparkClient", true, classloader)
        .asSubclass(classOf[SparkClient])
      val instance = cls.newInstance

      val conf = new SparkConf()
      conf.setAppName("AsakusaSparkClient")
      conf.setMaster("local[*]")

      val stageInfo = new StageInfo(
        sys.props("user.name"), "batchId", "flowId", null, "executionId", Map.empty[String, String])
      conf.setHadoopConf(Props.StageInfo, stageInfo.serialize)

      instance.execute(conf)
    } finally {
      Thread.currentThread.setContextClassLoader(cl)<commit_msg>Fix a test failure.<commit_after>package com.asakusafw.spark.compiler

import org.junit.runner.RunWith
import org.scalatest.FlatSpec
import org.scalatest.junit.JUnitRunner

import java.io.{ File, DataInput, DataOutput }
import java.net.URLClassLoader
import java.util.{ List => JList }

import scala.collection.JavaConversions._

import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{ NullWritable, Writable }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.SparkContext._

import com.asakusafw.bridge.stage.StageInfo
import com.asakusafw.lang.compiler.api.CompilerOptions
import com.asakusafw.lang.compiler.api.testing.MockJobflowProcessorContext
import com.asakusafw.lang.compiler.model.PropertyName
import com.asakusafw.lang.compiler.model.description.ClassDescription
import com.asakusafw.lang.compiler.model.graph._
import com.asakusafw.lang.compiler.model.info.ExternalInputInfo
import com.asakusafw.lang.compiler.model.testing.OperatorExtractor
import com.asakusafw.lang.compiler.planning._
import com.asakusafw.runtime.compatibility.JobCompatibility
import com.asakusafw.runtime.core.Result
import com.asakusafw.runtime.model.DataModel
import com.asakusafw.runtime.stage.StageConstants
import com.asakusafw.runtime.stage.input.TemporaryInputFormat
import com.asakusafw.runtime.stage.output.TemporaryOutputFormat
import com.asakusafw.runtime.value._
import com.asakusafw.spark.runtime._
import com.asakusafw.spark.tools.asm._
import com.asakusafw.vocabulary.flow.processor.PartialAggregation
import com.asakusafw.vocabulary.model.{ Key, Summarized }
import com.asakusafw.vocabulary.operator._

@RunWith(classOf[JUnitRunner])
class SparkClientCompilerSpecTest extends SparkClientCompilerSpec

class SparkClientCompilerSpec extends FlatSpec with LoadClassSugar {

  import SparkClientCompilerSpec._

  behavior of classOf[SparkClientCompiler].getSimpleName

  it should "compile Spark client from simple plan" in {
    val tmpDir = File.createTempFile("test-", null)
    tmpDir.delete
    val classpath = new File(tmpDir, "classes").getAbsoluteFile
    classpath.mkdirs()
    val path = new File(tmpDir, "tmp").getAbsolutePath

    spark { sc =>
      val hoges = sc.parallelize(0 until 100).map { i =>
        val hoge = new Hoge()
        hoge.id.modify(i)
        hoge
      }
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      job.setOutputKeyClass(classOf[NullWritable])
      job.setOutputValueClass(classOf[Hoge])
      job.setOutputFormatClass(classOf[TemporaryOutputFormat[Hoge]])
      TemporaryOutputFormat.setOutputPath(job, new Path(path, "extenal/input/hoge"))
      hoges.map((NullWritable.get, _)).saveAsNewAPIHadoopDataset(job.getConfiguration)
    }

    val inputOperator = ExternalInput
      .newInstance("hoge/part-*",
        new ExternalInputInfo.Basic(
          ClassDescription.of(classOf[Hoge]),
          "test",
          ClassDescription.of(classOf[Hoge]),
          ExternalInputInfo.DataSize.UNKNOWN))

    val outputOperator = ExternalOutput
      .newInstance("output", inputOperator.getOperatorPort)

    val graph = new OperatorGraph(Seq(inputOperator, outputOperator))

    val compiler = new SparkClientCompiler {

      override def preparePlan(graph: OperatorGraph): Plan = {
        val plan = super.preparePlan(graph)
        assert(plan.getElements.size === 2)
        plan
      }
    }

    val jpContext = new MockJobflowProcessorContext(
      new CompilerOptions("buildid", path, Map.empty[String, String]),
      Thread.currentThread.getContextClassLoader,
      classpath)
    val jobflow = new Jobflow("flowId", ClassDescription.of(classOf[SparkClientCompilerSpec]), graph)

    compiler.process(jpContext, jobflow)

    val cl = Thread.currentThread.getContextClassLoader
    try {
      val classloader = new URLClassLoader(Array(classpath.toURI.toURL), cl)
      Thread.currentThread.setContextClassLoader(classloader)
      val cls = Class.forName("com.asakusafw.generated.spark.flowId.SparkClient", true, classloader)
        .asSubclass(classOf[SparkClient])
      val instance = cls.newInstance

      val conf = new SparkConf()
      conf.setAppName("AsakusaSparkClient")
      conf.setMaster("local[*]")

      val stageInfo = new StageInfo(
        sys.props("user.name"), "batchId", "flowId", null, "executionId", Map.empty[String, String])
      conf.setHadoopConf(Props.StageInfo, stageInfo.serialize)

      instance.execute(conf)
    } finally {
      Thread.currentThread.setContextClassLoader(cl)
    }

    spark { sc =>
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      TemporaryInputFormat.setInputPaths(job, Seq(new Path(path, s"output/*/part-*")))
      val rdd = sc.newAPIHadoopRDD(
        job.getConfiguration,
        classOf[TemporaryInputFormat[Hoge]],
        classOf[NullWritable],
        classOf[Hoge])
      assert(rdd.map(_._2.id.get).collect === (0 until 100))
    }
  }

  it should "compile Spark client with Extract" in {
    val tmpDir = File.createTempFile("test-", null)
    tmpDir.delete
    val classpath = new File(tmpDir, "classes").getAbsoluteFile
    classpath.mkdirs()
    val path = new File(tmpDir, "tmp").getAbsolutePath

    spark { sc =>
      val hoges = sc.parallelize(0 until 100).map { i =>
        val hoge = new Hoge()
        hoge.id.modify(i)
        hoge
      }
      val job = JobCompatibility.newJob(sc.hadoopConfiguration)
      job.setOutputKeyClass(classOf[NullWritable])
      job.setOutputValueClass(classOf[Hoge])
      job.setOutputFormatClass(classOf[TemporaryOutputFormat[Hoge]])
      TemporaryOutputFormat.setOutputPath(job, new Path(path, "extenal/input/hoge"))
      hoges.map((NullWritable.get, _)).saveAsNewAPIHadoopDataset(job.getConfiguration)
    }

    val inputOperator = ExternalInput
      .newInstance("hoge/part-*",
        new ExternalInputInfo.Basic(
          ClassDescription.of(classOf[Hoge]),
          "test",
          ClassDescription.of(classOf[Hoge]),
          ExternalInputInfo.DataSize.UNKNOWN))

    val extractOperator = OperatorExtractor
      .extract(classOf[Extract], classOf[Ops], "extract")
      .input("hoge", ClassDescription.of(classOf[Hoge]), inputOperator.getOperatorPort)
      .output("evenResult", ClassDescription.of(classOf[Hoge]))
      .output("oddResult", ClassDescription.of(classOf[Hoge]))
      .build()

    val evenOutputOperator = ExternalOutput
      .newInstance("even", extractOperator.findOutput("evenResult"))

    val oddOutputOperator = ExternalOutput
      .newInstance("odd", extractOperator.findOutput("oddResult"))

    val graph = new OperatorGraph(Seq(
      inputOperator,
      extractOperator,
      evenOutputOperator,
      oddOutputOperator))

    val compiler = new SparkClientCompiler {

      override def preparePlan(graph: OperatorGraph): Plan = {
        val plan = super.preparePlan(graph)
        assert(plan.getElements.size === 3)
        plan
      }
    }

    val jpContext = new MockJobflowProcessorContext(
      new CompilerOptions("buildid", path, Map.empty[String, String]),
      Thread.currentThread.getContextClassLoader,
      classpath)
    val jobflow = new Jobflow("flowId", ClassDescription.of(classOf[SparkClientCompilerSpec]), graph)

    compiler.process(jpContext, jobflow)

    val cl = Thread.currentThread.getContextClassLoader
    try {
      val classloader = new URLClassLoader(Array(classpath.toURI.toURL), cl)
      Thread.currentThread.setContextClassLoader(classloader)
      val cls = Class.forName("com.asakusafw.generated.spark.flowId.SparkClient", true, classloader)
        .asSubclass(classOf[SparkClient])
      val instance = cls.newInstance

      val conf = new SparkConf()
      conf.setAppName("AsakusaSparkClient")
      conf.setMaster("local[*]")

      val stageInfo = new StageInfo(
        sys.props("user.name"), "batchId", "flowId", null, "executionId", Map.empty[String, String])
      conf.setHadoopConf(Props.StageInfo, stageInfo.serialize)

      instance.execute(conf)
    } finally {
      Thread.currentThread.setContextClassLoader(cl)
{'fraction_non_alphanumeric': 0.10587871287128713, 'fraction_numerical': 0.0018564356435643563, 'mean_word_length': 3.837174498653098, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7664530', 'n_tokens_mistral': 4998, 'n_tokens_neox': 4832, 'n_words': 854}