<commit_before>/**
 * Copyright (c) 2015 - The CM Authors <<EMAIL>>
 *   All Rights Reserved.
 *
 * This file is CONFIDENTIAL -- Distribution or duplication of this material or
 * the information contained herein is strictly forbidden unless prior written
 * permission is obtained.
 */
#include "stx/wallclock.h"
#include "stx/assets.h"
#include "stx/protobuf/msg.h"
#include "stx/io/BufferedOutputStream.h"
#include "zbase/api/EventsAPIServlet.h"

using namespace stx;

namespace zbase {

EventsAPIServlet::EventsAPIServlet(
    EventsService* service,
    ConfigDirectory* cdir,
    const String& cachedir) :
    service_(service),
    cdir_(cdir),
    cachedir_(cachedir) {}

void EventsAPIServlet::handle(
    const AnalyticsSession& session,
    RefPtr<stx::http::HTTPRequestStream> req_stream,
    RefPtr<stx::http::HTTPResponseStream> res_stream) {
  const auto& req = req_stream->request();
  URI uri(req.uri());

  http::HTTPResponse res;
  res.populateFromRequest(req);

  if (uri.path() == "/api/v1/events/scan") {
    scanTable(session, uri, req_stream.get(), res_stream.get());
    return;
  }

  if (uri.path() == "/api/v1/events/scan_partition") {
    scanTablePartition(session, uri, req_stream.get(), res_stream.get());
    return;
  }

  res.setStatus(http::kStatusNotFound);
  res.addHeader("Content-Type", "text/html; charset=utf-8");
  res.addBody(Assets::getAsset("zbase/webui/404.html"));
  res_stream->writeResponse(res);
}

void EventsAPIServlet::scanTable(
    const AnalyticsSession& session,
    const URI& uri,
    http::HTTPRequestStream* req_stream,
    http::HTTPResponseStream* res_stream) {
  req_stream->readBody();

  const auto& params = uri.queryParams();

  String table_name;
  if (!URI::getParam(params, "table", &table_name)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?table=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  EventScanParams scan_params;

  scan_params.set_end_time(WallClock::unixMicros());

  String time_str;
  if (URI::getParam(params, "time", &time_str)) {
    scan_params.set_end_time(std::stoull(time_str));
  }

  size_t limit = 1000;
  String limit_str;
  if (URI::getParam(params, "limit", &limit_str)) {
    limit = std::stoull(limit_str);
  }
  scan_params.set_limit(limit);

  String columns_str;
  if (URI::getParam(params, "columns", &columns_str)) {
    for (const auto& c : StringUtil::split(columns_str, ",")) {
      *scan_params.add_columns() = c;
    }
  }

  auto sse_stream = mkRef(new http::HTTPSSEStream (req_stream, res_stream));
  sse_stream->start();

  auto send_result_row = [sse_stream] (const msg::DynamicMessage& row) {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.addObjectEntry("event");
    row.toJSON(&json);
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("result"));
  };

  auto send_status_update = [sse_stream, scan_params] (bool done) {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.addObjectEntry("status");
    json.addString(done ? "finished" : "running");
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("progress"));
  };

  service_->scanTable(
      session,
      table_name,
      scan_params,
      send_result_row,
      send_status_update);

  sse_stream->finish();
}

void EventsAPIServlet::scanTablePartition(
    const AnalyticsSession& session,
    const URI& uri,
    http::HTTPRequestStream* req_stream,
    http::HTTPResponseStream* res_stream) {
  req_stream->readBody();

  const auto& params = uri.queryParams();

  String table;
  if (!URI::getParam(params, "table", &table)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?table=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  String partition;
  if (!URI::getParam(params, "partition", &partition)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?partition=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  String limit_str;
  if (!URI::getParam(params, "limit", &limit_str)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?limit=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  auto limit = std::stoull(limit_str);
  auto scan_params = msg::decode<EventScanParams>(
      req_stream->request().body());

  auto result = service_->scanLocalTablePartition(
      session,
      table,
      SHA1Hash::fromHexString(partition),
      scan_params);

  Buffer res_body;
  {
    BufferedOutputStream res_os(BufferOutputStream::fromBuffer(&res_body));
    result.encode(&res_os);
  }

  http::HTTPResponse res;
  res.populateFromRequest(req_stream->request());
  res.setStatus(http::kStatusOK);
  res.addBody(res_body);
  res_stream->writeResponse(res);
}

}
<commit_msg>TableScan: error event on error and query_completed event before finishing<commit_after>/**
 * Copyright (c) 2015 - The CM Authors <<EMAIL>>
 *   All Rights Reserved.
 *
 * This file is CONFIDENTIAL -- Distribution or duplication of this material or
 * the information contained herein is strictly forbidden unless prior written
 * permission is obtained.
 */
#include "stx/wallclock.h"
#include "stx/assets.h"
#include "stx/protobuf/msg.h"
#include "stx/io/BufferedOutputStream.h"
#include "zbase/api/EventsAPIServlet.h"

using namespace stx;

namespace zbase {

EventsAPIServlet::EventsAPIServlet(
    EventsService* service,
    ConfigDirectory* cdir,
    const String& cachedir) :
    service_(service),
    cdir_(cdir),
    cachedir_(cachedir) {}

void EventsAPIServlet::handle(
    const AnalyticsSession& session,
    RefPtr<stx::http::HTTPRequestStream> req_stream,
    RefPtr<stx::http::HTTPResponseStream> res_stream) {
  const auto& req = req_stream->request();
  URI uri(req.uri());

  http::HTTPResponse res;
  res.populateFromRequest(req);

  if (uri.path() == "/api/v1/events/scan") {
    scanTable(session, uri, req_stream.get(), res_stream.get());
    return;
  }

  if (uri.path() == "/api/v1/events/scan_partition") {
    scanTablePartition(session, uri, req_stream.get(), res_stream.get());
    return;
  }

  res.setStatus(http::kStatusNotFound);
  res.addHeader("Content-Type", "text/html; charset=utf-8");
  res.addBody(Assets::getAsset("zbase/webui/404.html"));
  res_stream->writeResponse(res);
}

void EventsAPIServlet::scanTable(
    const AnalyticsSession& session,
    const URI& uri,
    http::HTTPRequestStream* req_stream,
    http::HTTPResponseStream* res_stream) {
  req_stream->readBody();

  const auto& params = uri.queryParams();

  auto sse_stream = mkRef(new http::HTTPSSEStream (req_stream, res_stream));
  sse_stream->start();

  auto send_result_row = [sse_stream] (const msg::DynamicMessage& row) {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.addObjectEntry("event");
    row.toJSON(&json);
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("result"));
  };

  auto send_status_update = [sse_stream] (bool done) {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.addObjectEntry("status");
    json.addString(done ? "finished" : "running");
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("progress"));
  };

  try {
    String table_name;
    if (!URI::getParam(params, "table", &table_name)) {
      RAISE(kRuntimeError, "error: missing ?table=... parameter");
    }

    EventScanParams scan_params;
    scan_params.set_end_time(WallClock::unixMicros());

    String time_str;
    if (URI::getParam(params, "time", &time_str)) {
      scan_params.set_end_time(std::stoull(time_str));
    }

    size_t limit = 1000;
    String limit_str;
    if (URI::getParam(params, "limit", &limit_str)) {
      limit = std::stoull(limit_str);
    }
    scan_params.set_limit(limit);

    String columns_str;
    if (URI::getParam(params, "columns", &columns_str)) {
      for (const auto& c : StringUtil::split(columns_str, ",")) {
        *scan_params.add_columns() = c;
      }
    }

    service_->scanTable(
        session,
        table_name,
        scan_params,
        send_result_row,
        send_status_update);

  } catch (const StandardException& e) {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.addObjectEntry("error");
    json.addString(e.what());
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("query_error"));
    sse_stream->finish();
    return;
  }

  {
    Buffer buf;
    json::JSONOutputStream json(BufferOutputStream::fromBuffer(&buf));
    json.beginObject();
    json.endObject();

    sse_stream->sendEvent(buf, Some<String>("query_completed"));
  }

  sse_stream->finish();
}

void EventsAPIServlet::scanTablePartition(
    const AnalyticsSession& session,
    const URI& uri,
    http::HTTPRequestStream* req_stream,
    http::HTTPResponseStream* res_stream) {
  req_stream->readBody();

  const auto& params = uri.queryParams();

  String table;
  if (!URI::getParam(params, "table", &table)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?table=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  String partition;
  if (!URI::getParam(params, "partition", &partition)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?partition=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  String limit_str;
  if (!URI::getParam(params, "limit", &limit_str)) {
    http::HTTPResponse res;
    res.populateFromRequest(req_stream->request());
    res.setStatus(http::kStatusBadRequest);
    res.addBody("error: missing ?limit=... parameter");
    res_stream->writeResponse(res);
    return;
  }

  auto limit = std::stoull(limit_str);
  auto scan_params = msg::decode<EventScanParams>(
      req_stream->request().body());

  auto result = service_->scanLocalTablePartition(
      session,
      table,
      SHA1Hash::fromHexString(partition),
      scan_params);

  Buffer res_body;
  {
    BufferedOutputStream res_os(BufferOutputStream::fromBuffer(&res_body));
    result.encode(&res_os);
  }

  http::HTTPResponse res;
  res.populateFromRequest(req_stream->request());
  res.setStatus(http::kStatusOK);
  res.addBody(res_body);
  res_stream->writeResponse(res);
}

}

{'fraction_non_alphanumeric': 0.1457823683972094, 'fraction_numerical': 0.002718129926610492, 'mean_word_length': 4.309283309283309, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 62, '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': '1344932', 'n_tokens_mistral': 3622, 'n_tokens_neox': 3535, 'n_words': 642}