<commit_before>package latis.reader.tsml.agg

import scala.Option.option2Iterable
import latis.dm.Dataset
import latis.dm.Function
import latis.reader.tsml.TsmlAdapter
import latis.reader.tsml.TsmlReader
import latis.reader.tsml.ml.Tsml
import latis.util.iterator.PeekIterator
import latis.dm.Sample
import latis.util.FileUtils
import latis.dm.Text
import java.io.File

/**
 * An AggregationAdapter that reads data from each file in a file list 
 * and aggregates the results. The tsml should be structured like so: 
 * 
 * <dataset>
 *   <adapter class="latis.reader.tsml.agg.FileJoinAdapter"/>
 *   
 *   <dataset>
 *     ...
 *     This dataset returns the list of files to be read
 *     ...
 *   </dataset>
 *   
 *   <dataset>
 *     ...
 *     A dataset that gives the structure of the final dataset.
 *     For each file name in the list above, the template will be 
 *     used to read a dataset and the results will be aggregated 
 *     using the TileUnionAdapter.
 *     ...
 *   </dataset>
 *   
 * </dataset>
 *     
 */
class FileJoinAdapter(tsml: Tsml) extends TileUnionAdapter(tsml) {
  
  /**
   * Construct the adapter for the file list dataset.
   */
  override protected val adapters = List(TsmlAdapter(Tsml((tsml.xml \ "dataset").head)))
  
  /**
   * Tsml for the template to be used to read each file.
   */
  val template = Tsml((tsml.xml \ "dataset").last)
  
  /**
   * Extract the file paths from a file list Dataset.
   */
  def getFilePaths(ds: Dataset): Iterator[String] = {
    lazy val dir = ds.getMetadata.get("srcDir") match {
      case None => ""
      case Some(sd) => sd + File.separator
    }
    
    ds match {
      case Dataset(Function(it)) => it.map(_.findVariableByName("file") match {
        case Some(Text(file)) => dir + file
        case None => throw new Exception(s"No 'file' Variable found in Dataset '$ds'")
      })
    }
  }
  
  /**
   * Read each file and aggregate the results.
   */
  override def collect(datasets: Seq[Dataset]): Dataset = {
    //Get list of file names from the first dataset.
    val files = getFilePaths(datasets.head) 
    
    //Make a TsmlReader for each file from the tsml template with the file location inserted.
    val readers = files.map(file => TsmlReader(template.setLocation(file)))
    
    //Make an iterator over each file dataset, appending their samples
    val sit = readers.flatMap(r => r.getDataset match {
      case Dataset(Function(it)) => new PeekIterator[Sample] {
        def getNext = it.next match {
          case null => r.close; null; //TODO: need better assurance that readers get closed
          case sample => sample
        }
      }
    })
    
    val f = TsmlAdapter(template).getOrigDataset match {
      case Dataset(f: Function) => f //get the function template
    }
    
    val md = makeMetadata(tsml.dataset)
    Dataset(Function(f, sit), md)
  }

}
<commit_msg>don't apply projections to file list in FileJoinAdapter<commit_after>package latis.reader.tsml.agg

import scala.Option.option2Iterable
import latis.dm.Dataset
import latis.dm.Function
import latis.reader.tsml.TsmlAdapter
import latis.reader.tsml.TsmlReader
import latis.reader.tsml.ml.Tsml
import latis.util.iterator.PeekIterator
import latis.dm.Sample
import latis.util.FileUtils
import latis.dm.Text
import java.io.File
import latis.ops.Operation
import latis.ops.filter.Filter

/**
 * An AggregationAdapter that reads data from each file in a file list 
 * and aggregates the results. The tsml should be structured like so: 
 * 
 * <dataset>
 *   <adapter class="latis.reader.tsml.agg.FileJoinAdapter"/>
 *   
 *   <dataset>
 *     ...
 *     This dataset returns the list of files to be read
 *     ...
 *   </dataset>
 *   
 *   <dataset>
 *     ...
 *     A dataset that gives the structure of the final dataset.
 *     For each file name in the list above, the template will be 
 *     used to read a dataset and the results will be aggregated 
 *     using the TileUnionAdapter.
 *     ...
 *   </dataset>
 *   
 * </dataset>
 *     
 */
class FileJoinAdapter(tsml: Tsml) extends TileUnionAdapter(tsml) {
  
  /**
   * Construct the adapter for the file list dataset.
   */
  override protected val adapters = List(TsmlAdapter(Tsml((tsml.xml \ "dataset").head)))
  
  /**
   * Tsml for the template to be used to read each file.
   */
  val template = Tsml((tsml.xml \ "dataset").last)
  
  /**
   * Extract the file paths from a file list Dataset.
   */
  def getFilePaths(ds: Dataset): Iterator[String] = {
    lazy val dir = ds.getMetadata.get("srcDir") match {
      case None => ""
      case Some(sd) => sd + File.separator
    }
    
    ds match {
      case Dataset(Function(it)) => it.map(_.findVariableByName("file") match {
        case Some(Text(file)) => dir + file
        case None => throw new Exception(s"No 'file' Variable found in Dataset '$ds'")
      })
    }
  }
  
  /**
   * Override so that Projections aren't passed to the file list.
   */
  override def getDataset(ops: Seq[Operation]) = {
    val (filter, others) = ops.partition(_.isInstanceOf[Filter])
    val dss = adapters.map(_.getDataset(filter))
    
    val ds = collect(dss)
    
    ops.foldLeft(ds)((dataset, op) => op(dataset)) //doesn't handle any Operations
  }
  
  /**
   * Read each file and aggregate the results.
   */
  override def collect(datasets: Seq[Dataset]): Dataset = {
    //Get list of file names from the first dataset.
    val files = getFilePaths(datasets.head) 
    
    //Make a TsmlReader for each file from the tsml template with the file location inserted.
    val readers = files.map(file => TsmlReader(template.setLocation(file)))
    
    //Make an iterator over each file dataset, appending their samples
    val sit = readers.flatMap(r => r.getDataset match {
      case Dataset(Function(it)) => new PeekIterator[Sample] {
        def getNext = it.next match {
          case null => r.close; null; //TODO: need better assurance that readers get closed
          case sample => sample
        }
      }
    })
    
    val f = TsmlAdapter(template).getOrigDataset match {
      case Dataset(f: Function) => f //get the function template
    }
    
    val md = makeMetadata(tsml.dataset)
    Dataset(Function(f, sit), md)
  }

}

{'fraction_non_alphanumeric': 0.10990241561350184, 'fraction_numerical': 0.00031994880819068947, 'mean_word_length': 2.6841484973482617, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, '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': '2273481', 'n_tokens_mistral': 1942, 'n_tokens_neox': 1910, 'n_words': 600}