Method processOnTrigger

extensions/sql/processors/QueryDatabaseTable.cpp:324–368

}

void QueryDatabaseTable::processOnTrigger(core::ProcessSession& session) {
  const auto& selectQuery = getSelectQuery();

  logger_->log_info("QueryDatabaseTable: selectQuery: '%s'", selectQuery.c_str());

  auto statement = connection_->prepareStatement(selectQuery);

  auto rowset = statement->execute();

  int count = 0;
  size_t rowCount = 0;
  sql::MaxCollector maxCollector(selectQuery, maxValueColumnNames_, mapState_);
  sql::JSONSQLWriter sqlWriter(isJSONPretty());
  sql::SQLRowsetProcessor sqlRowsetProcessor(rowset, {&sqlWriter, &maxCollector});

  // Process rowset.
  do {
    rowCount = sqlRowsetProcessor.process(maxRowsPerFlowFile_ == 0 ? std::numeric_limits<size_t>::max() : maxRowsPerFlowFile_);
    count++;
    if (rowCount == 0)
      break;

    const auto output = sqlWriter.toString();
    if (!output.empty()) {
      WriteCallback writer(output);
      auto newflow = session.create();
      newflow->addAttribute(ResultRowCount, std::to_string(rowCount));
      newflow->addAttribute(ResultTableName, tableName_);
      session.write(newflow, &writer);
      session.transfer(newflow, s_success);
    }
  } while (rowCount > 0);

  const auto mapState = mapState_;
  if (maxCollector.updateMapState()) {
    try {
      session.commit();
    } catch (std::exception& e) {
      mapState_ = mapState;
      throw;
    }

    saveState();
  }
}

std::string QueryDatabaseTable::getSelectQuery() {
  std::string ret;