Method processOnSchedule

extensions/sql/processors/QueryDatabaseTable.cpp:229–322

}

void QueryDatabaseTable::processOnSchedule(core::ProcessContext &context) {
  initOutputFormat(context);

  context.getProperty(s_tableName.getName(), tableName_);
  context.getProperty(s_columnNames.getName(), columnNames_);

  context.getProperty(s_maxValueColumnNames.getName(), maxValueColumnNames_);
  listMaxValueColumnName_ = utils::inputStringToList(maxValueColumnNames_);

  context.getProperty(s_whereClause.getName(), whereClause_);
  context.getProperty(s_sqlQuery.getName(), sqlQuery_);
  context.getProperty(s_maxRowsPerFlowFile.getName(), maxRowsPerFlowFile_);

  mapState_.clear();

  state_manager_ = context.getStateManager();
  if (state_manager_ == nullptr) {
    throw Exception(PROCESSOR_EXCEPTION, "Failed to get StateManager");
  }

  std::unordered_map<std::string, std::string> state_map;
  if (state_manager_->get(state_map)) {
    if (state_map[TABLENAME_KEY] != tableName_) {
      state_manager_->clear();
    } else {
      for (auto&& elem : state_map) {
        if (elem.first.find(MAXVALUE_KEY_PREFIX) == 0) {
          mapState_.emplace(elem.first.substr(MAXVALUE_KEY_PREFIX.length()), std::move(elem.second));
        }
      }
    }
  } else {
    // Try to migrate legacy state file
    std::string stateDir;
    context.getProperty(s_stateDirectory.getName(), stateDir);
    if (!stateDir.empty()) {
      LegacyState legacyState(tableName_, stateDir, getUUIDStr(), logger_);
      if (legacyState) {
        mapState_ = legacyState.getStateMap();
        if (saveState() && state_manager_->persist()) {
          logger_->log_info("State migration successful");
          legacyState.moveStateFileToMigrated();
        } else {
          logger_->log_warn("Failed to persists migrated state");
        }
      } else {
        logger_->log_warn("Could not migrate state from specified State Directory %s", stateDir);
      }
    }
  }

  // If 'listMaxValueColumnName_' doesn't match columns in mapState_, then clear mapState_.
  if (listMaxValueColumnName_.size() != mapState_.size()) {
    mapState_.clear();
  } else {
    for (const auto& columName : listMaxValueColumnName_) {
      if (0 == mapState_.count(columName)) {
        mapState_.clear();