<commit_before>package org.vitrivr.adampro.utils

import java.io._

import com.google.protobuf.CodedInputStream
import io.grpc.stub.StreamObserver
import org.apache.commons.io.FileUtils
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types._
import org.vitrivr.adampro.datatypes.FieldTypes
import org.vitrivr.adampro.datatypes.FieldTypes.FieldType
import org.vitrivr.adampro.datatypes.feature.{FeatureVectorWrapper, FeatureVectorWrapperUDT}
import org.vitrivr.adampro.entity.Entity.EntityName
import org.vitrivr.adampro.entity.{AttributeDefinition, Entity}
import org.vitrivr.adampro.exception.GeneralAdamException
import org.vitrivr.adampro.grpc.grpc.InsertMessage.TupleInsertMessage
import org.vitrivr.adampro.grpc.grpc._
import org.vitrivr.adampro.main.AdamContext

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success, Try}

/**
  * ADAMpro
  *
  * <NAME>
  * November 2016
  */
class ProtoImporterExporter()(@transient implicit val ac: AdamContext) extends Serializable with Logging {
  private val BATCH_SIZE = 1000

  /**
    *
    * @param path
    * @param createOp
    * @param insertOp
    * @param observer
    */
  def importData(path: String, createOp: (CreateEntityMessage) => (Future[AckMessage]), insertOp: (InsertMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]) {
    if (!new File(path).exists()) {
      throw new GeneralAdamException("path does not exist")
    }

    readCatalogFile(getAllFiles(path).filter(_.getName.endsWith("catalog")), createOp, observer)
    readDataFile(getAllFiles(path).filter(_.getName.endsWith("bin")), insertOp, observer)
  }

  /**
    *
    * @param path
    * @return
    */
  private def getAllFiles(path: String) = {
    import scala.collection.JavaConverters._
    FileUtils.listFiles(new File(path), Array("bin", "catalog"), true).asScala.toList.sortBy(_.getAbsolutePath.reverse)
  }

  /**
    *
    * @param files
    * @param op
    * @param observer
    */
  private def readCatalogFile(files: Seq[File], op: (CreateEntityMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]): Unit = {
    val catalogfiles = files
    assert(catalogfiles.forall(_.getName.endsWith("catalog")))

    val batch = new ListBuffer[CreateEntityMessage]()

    log.trace("read catalog data")
    catalogfiles.foreach { path =>
      try {
        val is = new FileInputStream(path)

        try {
          val in = CodedInputStream.newInstance(is)

          while (!in.isAtEnd) {
            batch += CreateEntityMessage.parseDelimitedFrom(in).get
          }
        } catch {
          case e: Exception => log.error("exception while reading catalog file: " + path, e)
        }

        is.close()
      } catch {
        case e: Exception => log.error("exception while closing stream to catalog file: " + path, e)
      }
    }

    log.trace("perform creation of entities")
    batch.foreach { message =>
      try{
      op(message)
      } catch {
        case e : Exception => log.error("exception while creating entity " + message.entity, e)
      }
    }
  }

  /**
    *
    * @param files
    * @param op
    * @param observer
    */
  private def readDataFile(files: Seq[File], op: (InsertMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]): Unit = {
    val datafiles = files
    assert(datafiles.forall(_.getName.endsWith("bin")))

    val length = datafiles.length
    var done = 0

    log.info("will process " + length + " files")

    datafiles.grouped(BATCH_SIZE).foreach(pathBatch => {
      log.trace("starting new batch")
      val batch = new ListBuffer[InsertMessage]()

      pathBatch.foreach { path =>
        try {
          val entity = path.getName.replace(".bin", "")

          val is = new FileInputStream(path)

          try {
            val in = CodedInputStream.newInstance(is)

            while (!in.isAtEnd) {
              val tuple = TupleInsertMessage.parseDelimitedFrom(in).get

              val msg = InsertMessage(entity, Seq(tuple))

              batch += msg
            }
          } catch {
            case e: Exception => log.error("exception while reading files: " + path, e)
          }

          is.close()

          this.synchronized {
            done += 1
          }
        } catch {
          case e: Exception => log.error("exception while reading files: " + path, e)
        }
      }

      log.trace("inserting batch of length " + batch.length)<commit_msg>sending ack to client when errors while reading files<commit_after>package org.vitrivr.adampro.utils

import java.io._

import com.google.protobuf.CodedInputStream
import io.grpc.stub.StreamObserver
import org.apache.commons.io.FileUtils
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types._
import org.vitrivr.adampro.datatypes.FieldTypes
import org.vitrivr.adampro.datatypes.FieldTypes.FieldType
import org.vitrivr.adampro.datatypes.feature.{FeatureVectorWrapper, FeatureVectorWrapperUDT}
import org.vitrivr.adampro.entity.Entity.EntityName
import org.vitrivr.adampro.entity.{AttributeDefinition, Entity}
import org.vitrivr.adampro.exception.GeneralAdamException
import org.vitrivr.adampro.grpc.grpc.InsertMessage.TupleInsertMessage
import org.vitrivr.adampro.grpc.grpc._
import org.vitrivr.adampro.main.AdamContext

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success, Try}

/**
  * ADAMpro
  *
  * <NAME>
  * November 2016
  */
class ProtoImporterExporter()(@transient implicit val ac: AdamContext) extends Serializable with Logging {
  private val BATCH_SIZE = 1000

  /**
    *
    * @param path
    * @param createOp
    * @param insertOp
    * @param observer
    */
  def importData(path: String, createOp: (CreateEntityMessage) => (Future[AckMessage]), insertOp: (InsertMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]) {
    if (!new File(path).exists()) {
      throw new GeneralAdamException("path does not exist")
    }

    readCatalogFile(getAllFiles(path).filter(_.getName.endsWith("catalog")), createOp, observer)
    readDataFile(getAllFiles(path).filter(_.getName.endsWith("bin")), insertOp, observer)
  }

  /**
    *
    * @param path
    * @return
    */
  private def getAllFiles(path: String) = {
    import scala.collection.JavaConverters._
    FileUtils.listFiles(new File(path), Array("bin", "catalog"), true).asScala.toList.sortBy(_.getAbsolutePath.reverse)
  }

  /**
    *
    * @param files
    * @param op
    * @param observer
    */
  private def readCatalogFile(files: Seq[File], op: (CreateEntityMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]): Unit = {
    val catalogfiles = files
    assert(catalogfiles.forall(_.getName.endsWith("catalog")))

    val batch = new ListBuffer[CreateEntityMessage]()

    log.trace("read catalog data")
    catalogfiles.foreach { path =>
      try {
        val is = new FileInputStream(path)

        try {
          val in = CodedInputStream.newInstance(is)

          while (!in.isAtEnd) {
            batch += CreateEntityMessage.parseDelimitedFrom(in).get
          }
        } catch {
          case e: Exception => log.error("exception while reading catalog file: " + path, e)
        }

        is.close()
      } catch {
        case e: Exception => log.error("exception while closing stream to catalog file: " + path, e)
      }
    }

    log.trace("perform creation of entities")
    batch.foreach { message =>
      try{
      op(message)
      } catch {
        case e : Exception => log.error("exception while creating entity " + message.entity, e)
      }
    }
  }

  /**
    *
    * @param files
    * @param op
    * @param observer
    */
  private def readDataFile(files: Seq[File], op: (InsertMessage) => (Future[AckMessage]), observer: StreamObserver[AckMessage]): Unit = {
    val datafiles = files
    assert(datafiles.forall(_.getName.endsWith("bin")))

    val length = datafiles.length
    var done = 0

    log.info("will process " + length + " files")

    datafiles.grouped(BATCH_SIZE).foreach(pathBatch => {
      log.trace("starting new batch")
      val batch = new ListBuffer[InsertMessage]()

      pathBatch.foreach { path =>
        try {
          val entity = path.getName.replace(".bin", "")

          val is = new FileInputStream(path)

          try {
            val in = CodedInputStream.newInstance(is)

            while (!in.isAtEnd) {
              val tuple = TupleInsertMessage.parseDelimitedFrom(in).get

              val msg = InsertMessage(entity, Seq(tuple))

              batch += msg
            }
          } catch {
            observer.onNext(AckMessage(AckMessage.Code.ERROR, "error while reading files: " + path)))
            case e: Exception => log.error("exception while reading files: " + path, e)
          }

          is.close()

          this.synchronized {
            done += 1
          }
        } catch {
          observer.onNext(AckMessage(AckMessage.Code.ERROR, "exception while reading files: " + path)))
          case e: Exception => log.error("exception while reading files: " + path, e)
        }
      }

      log.trace("inserting batch of length " + batch.length)
{'fraction_non_alphanumeric': 0.11866581141757537, 'fraction_numerical': 0.0021381227282446015, 'mean_word_length': 2.886580805982551, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 31, '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': '1565862', 'n_tokens_mistral': 2853, 'n_tokens_neox': 2753, 'n_words': 595}