<commit_before>package com.socrata.soda.server.resources

import com.rojoma.json.v3.ast._
import com.rojoma.json.io.CompactJsonWriter
import com.rojoma.simplearm.util._
import com.rojoma.simplearm.v2.ResourceScope
import com.socrata.http.server.HttpRequest
import com.socrata.http.server.util.{NoPrecondition, RequestId}
import com.socrata.soda.clients.datacoordinator.DataCoordinatorClient._
import com.socrata.soda.server.responses.GeneralNotFoundError
import com.socrata.soda.server.{responses => SodaError, _}
import com.socrata.soda.server.computation.ComputedColumnsLike
import com.socrata.soda.server.export.JsonExporter
import com.socrata.soda.server.highlevel._
import com.socrata.soda.server.id.ResourceName
import com.socrata.soda.server.persistence._
import com.socrata.soql.environment.ColumnName
import javax.servlet.http.HttpServletResponse

class ComputeUtils(columnDAO: ColumnDAO, exportDAO: ExportDAO, rowDAO: RowDAO, computedColumns: ComputedColumnsLike) {

  val log = org.slf4j.LoggerFactory.getLogger(classOf[ComputeUtils])

  private def columnsToExport(requestId: RequestId.RequestId,
                              dataset: DatasetRecordLike,
                              computationStrategy: ComputationStrategyRecord): Seq[ColumnRecordLike] = {
    def sourceColumns    = getSourceColumns(requestId, dataset, computationStrategy)
    def primaryKeyColumn = dataset.columnsById(dataset.primaryKey)
    sourceColumns ++ Seq(primaryKeyColumn)
  }

  private def getSourceColumns(requestId: RequestId.RequestId,
                               dataset: DatasetRecordLike,
                               computationStrategy: ComputationStrategyRecord): Seq[ColumnRecordLike] = {
    computationStrategy.sourceColumns match {
      case Some(columns: Seq[MinimalColumnRecord]) =>
        val trans = new RowDataTranslator(requestId, dataset, false)
        trans.getInfoForColumnList(columns.map(_.id.underlying))
      case None => Seq()
    }
  }

  def compute(req: HttpRequest,
              response: HttpServletResponse,
              resourceScope: ResourceScope,
              resourceName: ResourceName,
              columnName: ColumnName,
              user: String)
             (successHandler: (HttpServletResponse, Iterator[ReportItem]) => Unit): Unit = {
    columnDAO.getColumn(resourceName, columnName) match {
      case ColumnDAO.Found(dataset, column, _)         =>
        column.computationStrategy match {
          case Some(strategy) => compute(req, response, resourceScope, dataset, column, user)(successHandler)
          case None           => SodaUtils.response(req, SodaError.NotAComputedColumn(columnName))(response)
        }
      case ColumnDAO.DatasetNotFound(dataset) =>
        SodaUtils.response(req, SodaError.DatasetNotFound(resourceName))(response)
      case ColumnDAO.ColumnNotFound(column)   =>
        SodaUtils.response(req, SodaError.ColumnNotFound(resourceName, columnName))(response)
      case _@x =>
        log.warn("case is NOT implemented")
        SodaUtils.response(req, GeneralNotFoundError("unexpected match case"))
    }
  }

  def compute(req: HttpRequest,
              response: HttpServletResponse,
              resourceScope: ResourceScope,
              dataset: DatasetRecordLike,
              column: ColumnRecordLike,
              user: String)
             (successHandler: (HttpServletResponse, Iterator[ReportItem]) => Unit): Unit =
    column.computationStrategy match {
      case Some(strategy) if computedColumns.computingEnabled(dataset.resourceName) =>
        val columns = columnsToExport(RequestId.getFromRequest(req), dataset, strategy)
        val requestId = RequestId.getFromRequest(req)
        log.info("export dataset {} for column compute", dataset.resourceName.name)
        val param = ExportParam(None, None, columns, None, sorted = false, rowId = None)
        exportDAO.export(dataset.resourceName,
                         NoPrecondition,
                         "latest",
                         param,
                         requestId,
                         resourceScope) match {
          case ExportDAO.Success(schema, newTag, rows) =>
            log.info("exported dataset {} for column compute", dataset.resourceName.name)
            val transformer = new RowDataTranslator(
              RequestId.getFromRequest(req), dataset, false)
            val upsertRows = transformer.transformDcRowsForUpsert(computedColumns, Seq(column), schema, rows)
            rowDAO.upsert(user, dataset, upsertRows, requestId)(UpsertUtils.handleUpsertErrors(req, response)(successHandler))
          case ExportDAO.SchemaInvalidForMimeType =>
            SodaUtils.response(req, SodaError.SchemaInvalidForMimeType)(response)
          case ExportDAO.NotModified(etags) =>
            SodaUtils.response(req, SodaError.ResourceNotModified(Nil, None))(response)
          case ExportDAO.PreconditionFailed =>
            SodaUtils.response(req, SodaError.EtagPreconditionFailed)(response)
          case ExportDAO.NotFound(resourceName) =>
            SodaUtils.response(req, SodaError.DatasetNotFound(resourceName))(response)
          case ExportDAO.InvalidRowId =>
            SodaUtils.response(req, SodaError.InvalidRowId)(response)
          case ExportDAO.InternalServerError(code, tag, data) =>
            SodaUtils.response(req, SodaError.InternalError(tag,
              "code"  -> JString(code),
              "data" -> JString(data)
            ))(response)
        }
      case Some(_) =>
        log.info("skipping computation for dataset {} for column compute", dataset.resourceName.name)
      case None =>
        SodaUtils.response(req, SodaError.NotAComputedColumn(column.fieldName))(response)
    }

  def writeComputeResponse(resourceName: ResourceName,
                           columnName: ColumnName,
                           responseCode: Int,
                           response: HttpServletResponse,
                           report: Iterator[ReportItem]) {
    import com.rojoma.json.ast._
    response.setStatus(responseCode)
    response.setContentType(SodaUtils.jsonContentTypeUtf8) // TODO: negotiate charset too
    using(response.getWriter) { w =>
      val jw = new CompactJsonWriter(w)
      var rowsComputed = 0
      while (report.hasNext) {
        report.next() match {
          case UpsertReportItem(items) =>
            // Data coordinator client throws an exception if we don't
            // iterate through the results.
            while(items.hasNext) {
              rowsComputed = rowsComputed + 1
              items.next()
            }<commit_msg>EN-17534: Fix bug in 73dfa0c6ecdfd5e1930b369e5a9dcdb9c04322a8<commit_after>package com.socrata.soda.server.resources

import com.rojoma.json.v3.ast._
import com.rojoma.json.io.CompactJsonWriter
import com.rojoma.simplearm.util._
import com.rojoma.simplearm.v2.ResourceScope
import com.socrata.http.server.HttpRequest
import com.socrata.http.server.util.{NoPrecondition, RequestId}
import com.socrata.soda.clients.datacoordinator.DataCoordinatorClient._
import com.socrata.soda.server.responses.GeneralNotFoundError
import com.socrata.soda.server.{responses => SodaError, _}
import com.socrata.soda.server.computation.ComputedColumnsLike
import com.socrata.soda.server.export.JsonExporter
import com.socrata.soda.server.highlevel._
import com.socrata.soda.server.id.ResourceName
import com.socrata.soda.server.persistence._
import com.socrata.soql.environment.ColumnName
import javax.servlet.http.HttpServletResponse

class ComputeUtils(columnDAO: ColumnDAO, exportDAO: ExportDAO, rowDAO: RowDAO, computedColumns: ComputedColumnsLike) {

  val log = org.slf4j.LoggerFactory.getLogger(classOf[ComputeUtils])

  private def columnsToExport(requestId: RequestId.RequestId,
                              dataset: DatasetRecordLike,
                              computationStrategy: ComputationStrategyRecord): Seq[ColumnRecordLike] = {
    def sourceColumns    = getSourceColumns(requestId, dataset, computationStrategy)
    def primaryKeyColumn = dataset.columnsById(dataset.primaryKey)
    sourceColumns ++ Seq(primaryKeyColumn)
  }

  private def getSourceColumns(requestId: RequestId.RequestId,
                               dataset: DatasetRecordLike,
                               computationStrategy: ComputationStrategyRecord): Seq[ColumnRecordLike] = {
    computationStrategy.sourceColumns match {
      case Some(columns: Seq[MinimalColumnRecord]) =>
        val trans = new RowDataTranslator(requestId, dataset, false)
        trans.getInfoForColumnList(columns.map(_.id.underlying))
      case None => Seq()
    }
  }

  def compute(req: HttpRequest,
              response: HttpServletResponse,
              resourceScope: ResourceScope,
              resourceName: ResourceName,
              columnName: ColumnName,
              user: String)
             (successHandler: (HttpServletResponse, Iterator[ReportItem]) => Unit): Unit = {
    columnDAO.getColumn(resourceName, columnName) match {
      case ColumnDAO.Found(dataset, column, _)         =>
        column.computationStrategy match {
          case Some(strategy) => compute(req, response, resourceScope, dataset, column, user)(successHandler)
          case None           => SodaUtils.response(req, SodaError.NotAComputedColumn(columnName))(response)
        }
      case ColumnDAO.DatasetNotFound(dataset) =>
        SodaUtils.response(req, SodaError.DatasetNotFound(resourceName))(response)
      case ColumnDAO.ColumnNotFound(column)   =>
        SodaUtils.response(req, SodaError.ColumnNotFound(resourceName, columnName))(response)
      case _@x =>
        log.warn("case is NOT implemented")
        SodaUtils.response(req, GeneralNotFoundError("unexpected match case"))
    }
  }

  def compute(req: HttpRequest,
              response: HttpServletResponse,
              resourceScope: ResourceScope,
              dataset: DatasetRecordLike,
              column: ColumnRecordLike,
              user: String)
             (successHandler: (HttpServletResponse, Iterator[ReportItem]) => Unit): Unit =
    column.computationStrategy match {
      case Some(strategy) if computedColumns.computingEnabled(dataset.resourceName) =>
        val columns = columnsToExport(RequestId.getFromRequest(req), dataset, strategy)
        val requestId = RequestId.getFromRequest(req)
        log.info("export dataset {} for column compute", dataset.resourceName.name)
        val param = ExportParam(None, None, columns, None, sorted = false, rowId = None)
        exportDAO.export(dataset.resourceName,
                         NoPrecondition,
                         "latest",
                         param,
                         requestId,
                         resourceScope) match {
          case ExportDAO.Success(schema, newTag, rows) =>
            log.info("exported dataset {} for column compute", dataset.resourceName.name)
            val transformer = new RowDataTranslator(
              RequestId.getFromRequest(req), dataset, false)
            val upsertRows = transformer.transformDcRowsForUpsert(computedColumns, Seq(column), schema, rows)
            rowDAO.upsert(user, dataset, upsertRows, requestId)(UpsertUtils.handleUpsertErrors(req, response)(successHandler))
          case ExportDAO.SchemaInvalidForMimeType =>
            SodaUtils.response(req, SodaError.SchemaInvalidForMimeType)(response)
          case ExportDAO.NotModified(etags) =>
            SodaUtils.response(req, SodaError.ResourceNotModified(Nil, None))(response)
          case ExportDAO.PreconditionFailed =>
            SodaUtils.response(req, SodaError.EtagPreconditionFailed)(response)
          case ExportDAO.NotFound(resourceName) =>
            SodaUtils.response(req, SodaError.DatasetNotFound(resourceName))(response)
          case ExportDAO.InvalidRowId =>
            SodaUtils.response(req, SodaError.InvalidRowId)(response)
          case ExportDAO.InternalServerError(code, tag, data) =>
            SodaUtils.response(req, SodaError.InternalError(tag,
              "code"  -> JString(code),
              "data" -> JString(data)
            ))(response)
        }
      case Some(_) =>
        log.info("skipping computation for dataset {} for column compute", dataset.resourceName.name)
        successHandler(response, Iterator.empty)
      case None =>
        SodaUtils.response(req, SodaError.NotAComputedColumn(column.fieldName))(response)
    }

  def writeComputeResponse(resourceName: ResourceName,
                           columnName: ColumnName,
                           responseCode: Int,
                           response: HttpServletResponse,
                           report: Iterator[ReportItem]) {
    import com.rojoma.json.ast._
    response.setStatus(responseCode)
    response.setContentType(SodaUtils.jsonContentTypeUtf8) // TODO: negotiate charset too
    using(response.getWriter) { w =>
      val jw = new CompactJsonWriter(w)
      var rowsComputed = 0
      while (report.hasNext) {
        report.next() match {
          case UpsertReportItem(items) =>
            // Data coordinator client throws an exception if we don't
            // iterate through the results.
            while(items.hasNext) {
              rowsComputed = rowsComputed + 1
              items.next()
            }
{'fraction_non_alphanumeric': 0.08996176624934403, 'fraction_numerical': 0.002848789264562561, 'mean_word_length': 2.7930054023315325, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 53, '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': '21404', 'n_tokens_mistral': 3621, 'n_tokens_neox': 3579, 'n_words': 664}