<commit_before>package io.eels.component.hive

import com.typesafe.config.ConfigFactory
import com.typesafe.scalalogging.slf4j.StrictLogging
import io.eels.HdfsIterator
import org.apache.hadoop.fs.{FileSystem, LocatedFileStatus, Path}
import org.apache.hadoop.hive.metastore.api.Table

object HiveFileExplorer extends StrictLogging {

  val config = ConfigFactory.load()
  val ignoreHiddenFiles = config.getBoolean("eel.hive.source.ignoreHiddenFiles")
  val hiddenFilePattern = config.getString("eel.hive.source.hiddenFilePattern")

  def apply(table: Table, partitionExprs: List[PartitionExpr])
           (implicit fs: FileSystem): List[Path] = {

    // returns true if the given file should be considered "hidden" based on the config settings
    def isHidden(file: LocatedFileStatus): Boolean = {
      ignoreHiddenFiles && file.getPath.getName.matches(hiddenFilePattern)
    }

    // returns true if the file meets the partition exprs
    def isEvaluated(file: LocatedFileStatus): Boolean = {
      // todo need better way of getting all partition info
      // todo this doesn't take into account partitions that are not stored under key=value semantics
      val partitions = Iterator.iterate(file.getPath.getParent)(path => path.getParent)
        .takeWhile(_ != null)
        .filter(_.getName.contains("="))
        .collect {
          case PartitionPart(name, value) => PartitionPart(name, value)
        }.toList
      partitionExprs.forall(_.eval(partitions))
    }

    // returns all files in the given location that are both visible and that match the partition expressions
    // todo this doesn't take into account partitions that are not stored under the table location
    def paths(location: String): List[Path] = {
      logger.debug(s"Scanning $location, filtering=$ignoreHiddenFiles, pattern=$hiddenFilePattern")
      val files = HdfsIterator(fs.listFiles(new Path(location), true)).filter(_.isFile).toList
      logger.debug(s"Found ${files.size} files before filtering")

      val visible = files.filterNot(isHidden).filter(isEvaluated).map(_.getPath)
      logger.info(s"Found ${visible.size} files after filtering")
      visible
    }

    val location = table.getSd.getLocation
    paths(location)
  }
}<commit_msg>Split hive file explorer into partitioned and non partitioned implementations<commit_after>package io.eels.component.hive

import com.typesafe.config.ConfigFactory
import com.typesafe.scalalogging.slf4j.StrictLogging
import io.eels.HdfsIterator
import org.apache.hadoop.fs.{FileSystem, LocatedFileStatus, Path}
import org.apache.hadoop.hive.metastore.api.Table

object HiveFileExplorer {
  def apply(table: Table, partitionExprs: List[PartitionExpr])(implicit fs: FileSystem): List[Path] = {
    if (partitionExprs.isEmpty) TableLocationHiveFileExplorer(table, partitionExprs)
    else PartitionHiveFileExplorer(table, partitionExprs)
  }
}

trait HiveFileExplorer extends StrictLogging {

  protected val config = ConfigFactory.load()
  protected val ignoreHiddenFiles = config.getBoolean("eel.hive.source.ignoreHiddenFiles")
  protected val hiddenFilePattern = config.getString("eel.hive.source.hiddenFilePattern")

  def apply(table: Table, partitionExprs: List[PartitionExpr])(implicit fs: FileSystem): List[Path]

  // returns true if the given file should be considered "hidden" based on the config settings
  def isHidden(file: LocatedFileStatus): Boolean = {
    ignoreHiddenFiles && file.getPath.getName.matches(hiddenFilePattern)
  }

  // returns all files in the given location that are visible
  def scan(location: String)(implicit fs: FileSystem): List[LocatedFileStatus] = {
    logger.debug(s"Scanning $location, filtering=$ignoreHiddenFiles, pattern=$hiddenFilePattern")
    val files = HdfsIterator(fs.listFiles(new Path(location), true)).filter(_.isFile).filterNot(isHidden).toList
    logger.debug(s"Found ${files.size} visible files")
    files
  }
}

// loads all files under the table location
object TableLocationHiveFileExplorer extends HiveFileExplorer {
  override def apply(table: Table, partitionExprs: List[PartitionExpr])(implicit fs: FileSystem): List[Path] = {
    val location = table.getSd.getLocation
    scan(location).map(_.getPath)
  }
}

object PartitionHiveFileExplorer extends HiveFileExplorer {

  override def apply(table: Table, partitionExprs: List[PartitionExpr])(implicit fs: FileSystem): List[Path] = {

    // returns true if the file meets the partition exprs
    def isEvaluated(file: LocatedFileStatus): Boolean = {
      // todo need better way of getting all partition info
      // todo this doesn't take into account partitions that are not stored under key=value semantics
      val partitions = Iterator.iterate(file.getPath.getParent)(path => path.getParent)
        .takeWhile(_ != null)
        .filter(_.getName.contains("="))
        .collect {
          case PartitionPart(name, value) => PartitionPart(name, value)
        }.toList
      partitionExprs.forall(_.eval(partitions))
    }

    val location = table.getSd.getLocation

    // returns all files in the given location that are both visible and that match the partition expressions
    // todo this doesn't take into account partitions that are not stored under the table location
    scan(location).filter(isEvaluated).map(_.getPath)
  }
}
{'fraction_non_alphanumeric': 0.09518434913468773, 'fraction_numerical': 0.0003762227238525207, 'mean_word_length': 4.311688311688312, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2277498', 'n_tokens_mistral': 1527, 'n_tokens_neox': 1474, 'n_words': 410}