// ---
// jupyter:
//   jupytext:
//     text_representation:
//       extension: .scala
//       format_name: light
//       format_version: '1.5'
//       jupytext_version: 1.14.4
//   kernelspec:
//     display_name: Scala 2.12
//     language: scala
//     name: scala212
// ---

val replClassPathObj = os.Path("replClasses", os.pwd)
if (!os.exists(replClassPathObj)) os.makeDir(replClassPathObj)
val replClassPath = replClassPathObj.toString()

interp.configureCompiler(_.settings.outputDirs.setSingleOutput(replClassPath))
interp.configureCompiler(_.settings.Yreplclassbased)
interp.load.cp(replClassPathObj)

// +
// Import the Snowpark library from Maven.
import $ivy.`com.snowflake:snowpark:1.1.0`
import $ivy.`org.apache.logging.log4j:log4j-core:2.17.1`

import com.snowflake.snowpark._
import com.snowflake.snowpark.functions._

import com.snowflake.snowpark.functions.{col, lit}
import com.snowflake.snowpark.{DataFrame, RelationalGroupedDataFrame, Session}

import java.io.{File, InputStream}
import java.nio.file.Paths
import java.util.Properties
import scala.collection.JavaConverters._
import scala.collection.mutable

// +

import org.apache.logging.log4j.core.config.Configurator;
import org.apache.logging.log4j.Level;
Configurator.setRootLevel(Level.WARN);


// -

    val session = Session.builder.configs(Map(
    "URL" -> "",
    "USER" -> "",
    "PASSWORD" -> "",
    "ROLE" -> "",
    "WAREHOUSE" -> "",
    "DB" -> "",
    "SCHEMA" -> ""
)).create

  def createDF(tableName : String) : DataFrame ={
      val df: DataFrame = session.table(tableName)
      df
  }

  def runProcess(tableName : String): Unit ={
    val df: DataFrame = session.table(tableName)
    val dfs = mutable.ArrayBuffer[DataFrame]()
    val fieldCount = df.schema.fields.size
    var iterCount = 0
    println(s"Total records in table : "
      + df.count().toString)
    df.schema.fields.foreach(i => {
      iterCount = iterCount + 1
      println(s"-- Processing column ${iterCount}/${fieldCount}: ${i.name}")
      println(s"Datatype : "+i.dataType.toString())
      val colAnalysis = createAnalysisDataframe(session,tableName, i.name, i.dataType.toString())
      dfs += colAnalysis
    })
    val analysisOutput = dfs.reduce(_ union _)
    analysisOutput.show(fieldCount)
    analysisOutput.write.mode("overwrite").saveAsTable(s"SNOWPARK_DATA_PROFILING_TPCDS_STORE")
  }


  // Rules definition
  def colValueDuplicates(df: DataFrame, columnName: String): String = {
    val dataframe1 = df.groupBy(columnName).count()
    val dataframe2 = dataframe1.filter(col("count") === 1)
    val duplicates = (dataframe1.count() - dataframe2.count()).toString
    println(s"No of duplicate records in column : ${duplicates}")
    duplicates
  }

  def colValueUniqueness(df: DataFrame, columnName: String): Int = {
    val dataframe1 = df.groupBy(col(columnName)).count()
    val dataframe2 = dataframe1.filter(col("count") === 1).count()

    println(s"No of unique values in column : "
      + dataframe2.toString)
    dataframe2.toInt
  }

  def colNullCheck(df: DataFrame, columnName: String): String = {
    val nulls = df.where(df.col(columnName).is_null).count()
    println(s"No of null values in column : "
      + nulls.toString)
    nulls.toString
  }

  def colValueCompleteness(df: DataFrame, columnName: String): String = {
    val nulls : Double = df.where(df.col(columnName).is_null).count()
    val count : Double = df.count().toDouble
    val completeness = 1-(nulls / count)
    println(s"Completeness of column : "
      + completeness.toString)
    completeness.toString
  }

  def colStatistics(df: DataFrame, columnName: String, dataType : String): String = {
    var stats = ""
    if (dataType != "String" && dataType != "Date") {
      val max = df.select(functions.max(df.col(columnName))).collect()(0)(0)
      val min = df.select(functions.min(df.col(columnName))).collect()(0)(0)
      val stdev = df.select(functions.stddev(df.col(columnName))).collect()(0)(0)
      val mean = df.select(functions.mean(df.col(columnName))).collect()(0)(0)
      stats = min + "/" + max + "/" + stdev + "/" + mean
      println(s"Statistics of column : " + stats)
      stats
    } else{
      stats
    }
  }

  def colDatatype(df: DataFrame, columnName: String): String = {
    val nulls : Double = df.where(df.col(columnName).is_null).count()
    //    val completeness = if (nulls>0) nulls/df.count() else 1.0
    val count : Double = df.count().toDouble
    val completeness = 1-(nulls / count)
    println(s"Completeness of column : " + completeness.toString)
    completeness.toString
  }

  def createAnalysisDataframe(session: Session, tableName : String, columnName : String, dataType : String): DataFrame = {
    val df: DataFrame = session.table(tableName)
    val dfSeq = session.createDataFrame(Seq(columnName)).toDF("columnName")
    val dfOut = dfSeq.withColumn("completeness", lit(colValueCompleteness(df, columnName)))
      .withColumn("uniqueValues", lit(colValueUniqueness(df, columnName)))
      .withColumn("duplicates",lit(colValueDuplicates(df, columnName)))
      .withColumn("nulls",lit(colNullCheck(df, columnName)))
      .withColumn("datatype",lit(dataType))
      .withColumn("MIN_MAX_MEAN_STDEV",lit(colStatistics(df, columnName, dataType)))
    dfOut
  }

  def createPermanentUdf(session: Session): Unit = {
    session.sql("drop FUNCTION if EXISTS udfPerm(Int)").show()
    session.udf.registerPermanent("udfPerm", permanentUdfHandler _, "STG")
    session.sql("select *, udfPerm(*) from values (10)").show()
  }

  def permanentUdfHandler(i: Int): Int = {
    i * i
  }

  def createInlinePermanentUdf(session: Session): Unit = {
    session.sql("drop FUNCTION if EXISTS udfInlinePerm(Int)").show()

    session.udf.registerPermanent("udfInlinePerm", (i: Int) => {
      i * 2
    }, "STG")

    session.sql("select *, udfInlinePerm(*) from values (10)").show()
  }

  private def addDeps(session: Session): Unit = {
    val PATH = Paths.get(".", "target", "dependency").toAbsolutePath.toString
    val lst = getListOfFiles(PATH)
    val filteredLst = lst.filterNot(_.matches("^.*snowpark(_original)?-[0-9.]+\\.jar$"))
    for (f <- filteredLst) {
      System.out.println("Adding dep:" + f)
      session.addDependency(f)
    }
  }

  def getListOfFiles(dir: String): List[String] = {
    val d = new File(dir)
    if (d.exists && d.isDirectory) {
      d.listFiles.filter(_.isFile).toList.map(_.getPath)
    } else {
      List[String]()
    }
  }
   

val df = createDF("SNOWFLAKE_SAMPLE_DATA.TPCDS_SF100TCL.STORE")
df.schema.fields.foreach(i => {println(i)})

runProcess("SNOWFLAKE_SAMPLE_DATA.TPCDS_SF100TCL.STORE")

StarcoderdataJupyterScriptsDedupFiltered