<commit_before>package filodb.repair

import java.io.File
import java.time.Instant
import java.time.format.DateTimeFormatter
import java.util

import com.typesafe.config.{Config, ConfigFactory}
import com.typesafe.scalalogging.StrictLogging
import monix.execution.Scheduler
import net.ceedubs.ficus.Ficus._
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import scala.concurrent.duration.{Duration, FiniteDuration}

import filodb.cassandra.FiloSessionProvider
import filodb.cassandra.columnstore.CassandraColumnStore
import filodb.core.{DatasetRef, GlobalConfig}
import filodb.core.store.ScanSplit
import filodb.downsampler.chunk.DownsamplerSettings


/**
  * Contains all the objects necessary for performing worker tasks. Is constructed from SparkConf
  * and is cached.
  */
class ChunkCopier(conf: SparkConf) {

  // Get filo config from spark conf or file.
  private def getFiloConfig(valueConf: String, filePathConf: String) = {
    val configString = conf.get(valueConf, "")
    val config = if (!configString.isBlank) {
      ConfigFactory.parseString(configString).resolve()
    } else {
      ConfigFactory.parseFile(new File(conf.get(filePathConf)))
        .withFallback(GlobalConfig.systemConfig)
        .resolve()
    }
    config
  }

  def datasetConfig(mainConfig: Config, datasetName: String): Config = {
    val sourceConfigPaths = mainConfig.getConfigList("inline-dataset-configs")
    sourceConfigPaths.stream()
      .filter(new util.function.Predicate[Config] {
        override def test(conf: Config): Boolean = conf.getString("dataset").equals(datasetName)
      })
      .findFirst()
      .orElseThrow()
  }

  val rawSourceConfig = getFiloConfig(
    "spark.filodb.chunks.copier.source.config.value",
    "spark.filodb.chunks.copier.source.config.file"
  )
  val rawTargetConfig = getFiloConfig(
    "spark.filodb.chunks.copier.target.config.value",
    "spark.filodb.chunks.copier.target.config.file"
  )
  val sourceConfig = rawSourceConfig.getConfig("filodb")
  val targetConfig = rawTargetConfig.getConfig("filodb")
  val sourceCassConfig = sourceConfig.getConfig("cassandra")
  val targetCassConfig = targetConfig.getConfig("cassandra")

  val isDownsampleCopy = conf.getBoolean("spark.filodb.chunks.copier.is.downsample.copy", false)

  val datasetName = conf.get("spark.filodb.chunks.copier.dataset")
  val targetDatasetConfig = datasetConfig(targetConfig, datasetName)
  val datasetRef = if (isDownsampleCopy) {
    val downsampleResolution = Duration(conf.get("spark.filodb.chunks.copier.dataset.downsample.resolution"))
    DatasetRef(s"${datasetName}_ds_${downsampleResolution.toMinutes}")
  } else {
    DatasetRef.fromDotString(datasetName)
  }

  // Examples: 2019-10-20T12:34:56Z  or  2019-10-20T12:34:56-08:00
  private def parseDateTime(str: String) = Instant.from(DateTimeFormatter.ISO_OFFSET_DATE_TIME.parse(str))

  val copyStartTime = parseDateTime(conf.get("spark.filodb.chunks.copier.start.time"))
  val copyEndTime = parseDateTime(conf.get("spark.filodb.chunks.copier.end.time"))

  // Destructively deletes everything in the target before updating anthing. Is used when chunks aren't aligned.
  val deleteFirst = conf.getBoolean("spark.filodb.chunks.copier.delete.first", false)

  // Disable the copy phase either for fully deleting with no replacement, or for no-op testing.
  val noCopy = conf.getBoolean("spark.filodb.chunks.copier.noCopy", false)

  val diskTimeToLiveSeconds = if (isDownsampleCopy) {
    val dsSettings = new DownsamplerSettings(rawSourceConfig)
    val highestDSResolution = dsSettings.rawDatasetIngestionConfig.downsampleConfig.resolutions.last
    dsSettings.ttlByResolution(highestDSResolution)
  } else {
    targetDatasetConfig.getConfig("sourceconfig.store")
      .as[FiniteDuration]("disk-time-to-live").toSeconds.toInt
  }

  val numSplitsForScans = sourceCassConfig.getInt("num-token-range-splits-for-scans")
  val batchSize = conf.getInt("spark.filodb.chunks.copier.batchSize", 10000)

  val readSched = Scheduler.io("cass-read-sched")
  val writeSched = Scheduler.io("cass-write-sched")
<commit_msg>bug(repair): Use resolution specific ttl for downsample dataset in chunks repair. (#1181)<commit_after>package filodb.repair

import java.io.File
import java.time.Instant
import java.time.format.DateTimeFormatter
import java.util

import com.typesafe.config.{Config, ConfigFactory}
import com.typesafe.scalalogging.StrictLogging
import monix.execution.Scheduler
import net.ceedubs.ficus.Ficus._
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import scala.concurrent.duration.{Duration, FiniteDuration}

import filodb.cassandra.FiloSessionProvider
import filodb.cassandra.columnstore.CassandraColumnStore
import filodb.core.{DatasetRef, GlobalConfig}
import filodb.core.store.ScanSplit
import filodb.downsampler.chunk.DownsamplerSettings


/**
  * Contains all the objects necessary for performing worker tasks. Is constructed from SparkConf
  * and is cached.
  */
class ChunkCopier(conf: SparkConf) {

  // Get filo config from spark conf or file.
  private def getFiloConfig(valueConf: String, filePathConf: String) = {
    val configString = conf.get(valueConf, "")
    val config = if (!configString.isBlank) {
      ConfigFactory.parseString(configString).resolve()
    } else {
      ConfigFactory.parseFile(new File(conf.get(filePathConf)))
        .withFallback(GlobalConfig.systemConfig)
        .resolve()
    }
    config
  }

  def datasetConfig(mainConfig: Config, datasetName: String): Config = {
    val sourceConfigPaths = mainConfig.getConfigList("inline-dataset-configs")
    sourceConfigPaths.stream()
      .filter(new util.function.Predicate[Config] {
        override def test(conf: Config): Boolean = conf.getString("dataset").equals(datasetName)
      })
      .findFirst()
      .orElseThrow()
  }

  val rawSourceConfig = getFiloConfig(
    "spark.filodb.chunks.copier.source.config.value",
    "spark.filodb.chunks.copier.source.config.file"
  )
  val rawTargetConfig = getFiloConfig(
    "spark.filodb.chunks.copier.target.config.value",
    "spark.filodb.chunks.copier.target.config.file"
  )
  val sourceConfig = rawSourceConfig.getConfig("filodb")
  val targetConfig = rawTargetConfig.getConfig("filodb")
  val sourceCassConfig = sourceConfig.getConfig("cassandra")
  val targetCassConfig = targetConfig.getConfig("cassandra")

  val isDownsampleCopy = conf.getBoolean("spark.filodb.chunks.copier.is.downsample.copy", false)

  val datasetName = conf.get("spark.filodb.chunks.copier.dataset")
  val targetDatasetConfig = datasetConfig(targetConfig, datasetName)
  val datasetRef = if (isDownsampleCopy) {
    val downsampleResolution = Duration(conf.get("spark.filodb.chunks.copier.dataset.downsample.resolution"))
    DatasetRef(s"${datasetName}_ds_${downsampleResolution.toMinutes}")
  } else {
    DatasetRef.fromDotString(datasetName)
  }

  // Examples: 2019-10-20T12:34:56Z  or  2019-10-20T12:34:56-08:00
  private def parseDateTime(str: String) = Instant.from(DateTimeFormatter.ISO_OFFSET_DATE_TIME.parse(str))

  val copyStartTime = parseDateTime(conf.get("spark.filodb.chunks.copier.start.time"))
  val copyEndTime = parseDateTime(conf.get("spark.filodb.chunks.copier.end.time"))

  // Destructively deletes everything in the target before updating anthing. Is used when chunks aren't aligned.
  val deleteFirst = conf.getBoolean("spark.filodb.chunks.copier.delete.first", false)

  // Disable the copy phase either for fully deleting with no replacement, or for no-op testing.
  val noCopy = conf.getBoolean("spark.filodb.chunks.copier.noCopy", false)

  val diskTimeToLiveSeconds = if (isDownsampleCopy) {
    val dsSettings = new DownsamplerSettings(rawSourceConfig)
    val downsampleResolution = Duration(conf.get("spark.filodb.chunks.copier.dataset.downsample.resolution"))
      .asInstanceOf[FiniteDuration]
    dsSettings.ttlByResolution(downsampleResolution)
  } else {
    targetDatasetConfig.getConfig("sourceconfig.store")
      .as[FiniteDuration]("disk-time-to-live").toSeconds.toInt
  }

  val numSplitsForScans = sourceCassConfig.getInt("num-token-range-splits-for-scans")
  val batchSize = conf.getInt("spark.filodb.chunks.copier.batchSize", 10000)

  val readSched = Scheduler.io("cass-read-sched")
  val writeSched = Scheduler.io("cass-write-sched")

{'fraction_non_alphanumeric': 0.10952836964328465, 'fraction_numerical': 0.009336844625329183, 'mean_word_length': 4.794036061026352, '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': '6984525', 'n_tokens_mistral': 2762, 'n_tokens_neox': 2584, 'n_words': 468}