<commit_before>package io.spark.ddf.etl

import org.apache.spark.sql.DataFrame

import scala.collection.JavaConverters._
import scala.collection.JavaConversions.asScalaIterator
import scala.collection.JavaConversions.seqAsJavaList

import org.apache.spark.SparkContext.rddToPairRDDFunctions
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.rosuda.REngine.REXP
import org.rosuda.REngine.REXPDouble
import org.rosuda.REngine.REXPInteger
import org.rosuda.REngine.REXPList
import org.rosuda.REngine.REXPLogical
import org.rosuda.REngine.REXPString
import org.rosuda.REngine.RList
import org.rosuda.REngine.Rserve.RConnection
import org.rosuda.REngine.Rserve.StartRserve

import io.ddf.DDF
import io.ddf.content.Schema
import io.ddf.content.Schema.Column
import io.ddf.etl.{TransformationHandler ⇒ CoreTransformationHandler}
import io.ddf.exception.DDFException
import io.spark.ddf.util.SparkUtils
import java.util.{ArrayList, List}

class TransformationHandler(mDDF: DDF) extends CoreTransformationHandler(mDDF) {


  override def flattenDDF(selectedColumns: List[String]): DDF = {
    val df: DataFrame = mDDF.getRepresentationHandler.get(classOf[DataFrame]).asInstanceOf[DataFrame]
    val flattenedColumns: ArrayList[String] = SparkUtils.flattenColumnNamesFromDataFrame(df, selectedColumns)

    val selectClause = flattenedColumns.asScala.toList.mkString(",")
    val q = String.format("select %s from %s", selectClause, mDDF.getTableName)

    mDDF.sql2ddf(q)
  }



  override def transformMapReduceNative(mapFuncDef: String, reduceFuncDef: String, mapsideCombine: Boolean = true): DDF = {

    // Prepare data as REXP objects
    val dfrdd = mDDF.getRepresentationHandler.get(classOf[RDD[_]], classOf[REXP]).asInstanceOf[RDD[REXP]]

    // 1. map!
    val rMapped = dfrdd.map {
      partdf ⇒
        try {
          TransformationHandler.preShuffleMapper(partdf, mapFuncDef, reduceFuncDef, mapsideCombine)
        } catch {
          case e: Exception ⇒ {

            e match {
              case aExc: DDFException ⇒ throw aExc
              case rserveExc: org.rosuda.REngine.Rserve.RserveException ⇒ {
                throw new DDFException(rserveExc.getMessage, null)
              }
              case _ ⇒ throw new DDFException(e.getMessage, null)
            }
          }<commit_msg>add API with empty parameter for flattenDDF<commit_after>package io.spark.ddf.etl

import org.apache.spark.sql.DataFrame

import scala.collection.JavaConverters._
import scala.collection.JavaConversions.asScalaIterator
import scala.collection.JavaConversions.seqAsJavaList

import org.apache.spark.SparkContext.rddToPairRDDFunctions
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.rosuda.REngine.REXP
import org.rosuda.REngine.REXPDouble
import org.rosuda.REngine.REXPInteger
import org.rosuda.REngine.REXPList
import org.rosuda.REngine.REXPLogical
import org.rosuda.REngine.REXPString
import org.rosuda.REngine.RList
import org.rosuda.REngine.Rserve.RConnection
import org.rosuda.REngine.Rserve.StartRserve

import io.ddf.DDF
import io.ddf.content.Schema
import io.ddf.content.Schema.Column
import io.ddf.etl.{TransformationHandler ⇒ CoreTransformationHandler}
import io.ddf.exception.DDFException
import io.spark.ddf.util.SparkUtils
import java.util.{ArrayList, List}

class TransformationHandler(mDDF: DDF) extends CoreTransformationHandler(mDDF) {


  override def flattenDDF(selectedColumns: List[String]): DDF = {
    val df: DataFrame = mDDF.getRepresentationHandler.get(classOf[DataFrame]).asInstanceOf[DataFrame]
    val flattenedColumns: ArrayList[String] = SparkUtils.flattenColumnNamesFromDataFrame(df, selectedColumns)

    val selectClause = flattenedColumns.asScala.toList.mkString(",")
    val q = String.format("select %s from %s", selectClause, mDDF.getTableName)

    mDDF.sql2ddf(q)
  }

  override def flattenDDF(): DDF = {
    flattenDDF(null)
  }

  override def transformMapReduceNative(mapFuncDef: String, reduceFuncDef: String, mapsideCombine: Boolean = true): DDF = {

    // Prepare data as REXP objects
    val dfrdd = mDDF.getRepresentationHandler.get(classOf[RDD[_]], classOf[REXP]).asInstanceOf[RDD[REXP]]

    // 1. map!
    val rMapped = dfrdd.map {
      partdf ⇒
        try {
          TransformationHandler.preShuffleMapper(partdf, mapFuncDef, reduceFuncDef, mapsideCombine)
        } catch {
          case e: Exception ⇒ {

            e match {
              case aExc: DDFException ⇒ throw aExc
              case rserveExc: org.rosuda.REngine.Rserve.RserveException ⇒ {
                throw new DDFException(rserveExc.getMessage, null)
              }
              case _ ⇒ throw new DDFException(e.getMessage, null)
            }
          }
{'fraction_non_alphanumeric': 0.09237875288683603, 'fraction_numerical': 0.000839806844425782, 'mean_word_length': 3.911340206185567, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5217195', 'n_tokens_mistral': 1606, 'n_tokens_neox': 1565, 'n_words': 291}