grid_processor.cpp

Go to the documentation of this file.

/*****************************************************************************\
*                                                                             *
*  Name   : grid_processor                                                    *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 2008-$now By Author.  This program is free software; you can  *
* redistribute it and/or modify it under the terms of the GNU General Public  *
* License as published by the Free Software Foundation; either version 2 of   *
* the License or (at your option) any later version.  This is online at:      *
*     http://www.fsf.org/copyleft/gpl.html                                    *
* Please send any updates to: fred@gruntose.com                               *
\*****************************************************************************/

#include "common.h"
#include "grid_processor.h"
#include "login_info.h"

#include <basis/function.h>
#include <basis/log_base.h>
#include <basis/portable.h>
#include <basis/set.cpp>
#include <basis/string_array.h>
#include <data_struct/deep_array.cpp>
#include <mechanisms/time_stamp.cpp>
#include <opsystem/byte_filer.h>
#include <opsystem/directory.h>
#include <opsystem/filename.h>
#include <opsystem/ini_config.h>
#include <textual/list_parsing.h>

#include <malloc.h>

#define LOG(s) CLASS_EMERGENCY_LOG(program_wide_logger(), filename(portable::application_name()).basename().raw() + " " + s)

const int GRID_CHECK_PAUSE = int(2.5 * MINUTE_ms);


class float_array : public array<float> {};


grid_processor::grid_processor(query_handler &work_with)
: _querier(work_with),
  _startup_time(new time_stamp),
  _complain_on_missing_grid(true)
{
}

grid_processor::~grid_processor()
{
  WHACK(_startup_time);
}

bool grid_processor::write_bogus_row(byte_filer &df_file, int data_source,
   int row_number)
{
  FUNCDEF("write_bogus_row");
  DBPROCESS *current_db = _querier.current_db();
  if (!current_db) { LOG("ERROR: null database object"); return false; }

//LOG("got to fudging rows now...");

  bool to_return = true;
  for (int i = 1; i <= dbnumcols(current_db); i++) {
    string_array items;
    // add the datafield id.
    items += isprintf("%d", row_number * 100 + i);

    // add the status.
    items += "UNAVAILABLE";
    int type = dbcoltype(current_db, i);
    switch (type) {
      case SYBINT1:
      case SYBINT2:
      case SYBINT4: {
        items += "INT32";
        // multiply by the factor we were given to calculate the final value.
        items += "-1";
        break;
      }
      case SYBCHAR:
      case SYBVARCHAR:
      case SYBTEXT: {
//STRINGBIND
//VARYCHARBIND
        items += "STRING";
        items += "blank";
        break;
      }
      default: {
        items += "STRING";
        items += "unhandled";
        break;
      }
    }

    // create this line as a CSV formatted string and put it in the file.
    istring output_line;
    list_parsing::create_csv_line(items, output_line);
    output_line += log_base::platform_ending();
    df_file.write(output_line);
  }

  return to_return;
}

bool grid_processor::write_fields_for_row(byte_filer &df_file, int data_source,
    const istring &key_column, const string_array &row_names,
    float_array &multipliers, int_set &rows_seen)
{
  FUNCDEF("write_fields_for_row");
  DBPROCESS *current_db = _querier.current_db();
  if (!current_db) { LOG("ERROR: null database object"); return false; }

  int current_row = -1;  // until we know which row it is.

//LOG(isprintf("got numcols = %d", dbnumcols(current_db)));

  // first figure out which freaking row this is.
  for (int i = 1; i <= dbnumcols(current_db); i++) {
    if (current_row != -1) break;  // found a match already.
    istring cur_col_name = dbcolname(current_db, i);
    if (cur_col_name == key_column) {
      // this is the right column; it's the key.  now compare its values.
//LOG(isprintf("found column same as key at col# %d", i));
      istring found_at_i = _querier.get_bound_string(i);
      // handle a key that's an integer column.
      if ( (dbcoltype(current_db, i) == SYBINT1)
          || (dbcoltype(current_db, i) == SYBINT2)
          || (dbcoltype(current_db, i) == SYBINT4) ) {
        found_at_i = isprintf("%d", _querier.get_bound_int(i));
      }

      // search through the rows and see if this is an expected row.
      for (int j = 0; j < row_names.length(); j++) {
        if (found_at_i.iequals(row_names[j])) {
          // found the right row name for the row we got from DB.
          current_row = j;
          break;
        }
      }
//LOG("breaking out due to having found correct column");
      // we found the right column already, so stop looking.
      break;
    }
  }

//LOG(isprintf("db num cols = %d", dbnumcols(current_db)));

  if (negative(current_row)) {
    // we never found the column; complain.
    LOG(istring("ERROR: could not find column '") + key_column
        + "' to determine what row this is.  columns are:");
    istring show_cols;
    for (int i = 1; i <= dbnumcols(current_db); i++) {
      if (i > 1) show_cols += "\t";
      show_cols += dbcolname(current_db, i);
    }
    LOG(show_cols);
    return false;
  }

//LOG("got to scanning rows now...");

  // we know what row we *think* it's at, but if there were multiple of
  // the same row, we need to fill in all the lines with the one match.
  istring matching_row_name = row_names[current_row];
  int_set rows_to_fill;
  for (int j = 0; j < row_names.length(); j++) {
    // at least one of these should be our original match.
    if (matching_row_name.iequals(row_names[j]))
      rows_to_fill += j;
  }

  rows_seen += rows_to_fill;  // remember which ones we already hit.

  bool to_return = true;
  for (int fill_which = 0; fill_which < rows_to_fill.length(); fill_which++) {
    // get the next row # on display that gets this line of data.
    int row_to_store = rows_to_fill[fill_which];
    for (int i = 1; i <= dbnumcols(current_db); i++) {
      string_array items;
      // add the datafield id.
      items += isprintf("%d", row_to_store * 100 + i);

      // add the status.
      items += "OK";
      int type = dbcoltype(current_db, i);
      switch (type) {
        case SYBINT1:
        case SYBINT2:
        case SYBINT4: {
          items += "INT32";
          // multiply by the factor we were given to calculate the final value.
          int new_value = int(multipliers[i - 1] * _querier.get_bound_int(i));
          items += isprintf("%d", new_value);
          break;
        }
        case SYBCHAR:
        case SYBVARCHAR:
        case SYBTEXT: {
//STRINGBIND
//VARYCHARBIND
          items += "STRING";
          istring found = _querier.get_bound_string(i);
          found.shrink();  // get proper length.
          items += found;
          break;
        }
        default: {
          items += "STRING";
          items += "unhandled";
          break;
        }
      }

      // create this line as a CSV formatted string and put it in the file.
      istring output_line;
      list_parsing::create_csv_line(items, output_line);
      output_line += log_base::platform_ending();
      df_file.write(output_line);
    }
  }
  return to_return;
}

void grid_processor::write_data_fields(int data_source,
    const istring &df_storage_dir, const istring &key_column,
    const string_array &row_names, float_array &multipliers)
{
  FUNCDEF("write_data_fields");
  DBPROCESS *current_db = _querier.current_db();
  if (!current_db) { LOG("ERROR: null database object"); return; }
  int db_result = REG_ROW;

  bool bind_worked = _querier.bind_columns();
  if (!bind_worked) {
//hmmm: bail out when the result comes back other than success.
    return;
  }

  // open up our output file where we will dump all the result data.
  byte_filer df_file(istring(df_storage_dir)
      + isprintf("/df_%d.csv", data_source), "wb");

  int_set rows_seen;  // tracks which rows were handled.

  while (db_result == REG_ROW) {
    // see whether there's a next row.
    db_result = dbnextrow(current_db);
    if (db_result == REG_ROW) {
      write_fields_for_row(df_file, data_source, key_column, row_names,
          multipliers, rows_seen);
    } else if (db_result != NO_MORE_ROWS) {
      // report that an unexpected error occurred.      
      LOG(isprintf("ERROR: failure to get next row, error code is %d", db_result));
    }
  }

  int_set all_rows_expected;
  for (int i = 0; i < row_names.length(); i++) {
    all_rows_expected += i;
  }
  int_set missing_rows = all_rows_expected - rows_seen;
  for (int i = 0; i < missing_rows.length(); i++) {
    int one_row_missing = missing_rows[i];
    write_bogus_row(df_file, data_source, one_row_missing);
  }

  df_file.close();
}

bool grid_processor::process_query_as_df(const istring &query, int data_source,
    const istring &df_storage_dir, const istring &key_column,
    const string_array &row_names, float_array &multipliers)
{
  FUNCDEF("process_query_as_df");
  DBPROCESS *current_db = _querier.current_db();
  if (!current_db) { LOG("ERROR: null database object"); return false; }
//LOG(istring("process_query_as_df on: ") + query);
  dbfreebuf(current_db);
  dbcmd(current_db, query.s());
  dbsqlexec(current_db);

  if (dbresults(current_db) != SUCCEED) {
    LOG(isprintf("ERROR: was expecting a result set and got none."));
    return false;
  }
  write_data_fields(data_source, df_storage_dir, key_column,
      row_names, multipliers);
  return true;
}

void grid_processor::reset_data_fields(const istring &df_storage_dir,
    int data_source)
{
  FUNCDEF("reset_data_fields");
  byte_filer df_file(istring(df_storage_dir)
      + isprintf("/df_%d.csv", data_source), "rb");
  string_array lines;
  while (!df_file.eof()) {
    istring line;
    int chars = df_file.getline(line, 2048);
      // ridiculous but sufficient limit.
    if (chars > 0) {
      lines += line;
    }
  }
  df_file.close();
  df_file.open(istring(df_storage_dir)
      + isprintf("/df_%d.csv", data_source), "wb");
  for (int i = 0; i < lines.length(); i++) {
    istring line = lines[i];
    int comma_indy = line.find(',', 0);
    if (negative(comma_indy)) {
      LOG(istring("ERROR: failed to parse line: ") + line);
      return;  // hose-up.  stop.
    }
    int start_word = comma_indy + 2;  // skip comma and quote.
    istring word_accum;
    int last_word_indy = -1;
    for (int j = start_word; j < line.length(); j++) {
      char curr = line[j];
      if ( (curr != ',') && (curr != '"') ) {
        word_accum += istring(curr, 1);
        last_word_indy = j;
      } else break;  // done looking.
    }
    if (!word_accum.iequals("unavailable")) {
      // take out the wrong word and replace it.
      istring new_line = line.substring(0, start_word - 1);
      new_line += "UNAVAILABLE";
      new_line += line.substring(last_word_indy + 1, line.end());
      df_file.write(new_line);
    } else {
      // no processing on the line.
      df_file.write(line);
    }
  }
  df_file.close();
}

void grid_processor::process_grid_files(const istring &grid_storage_folder,
    const istring &df_storage_dir, database_login_info &persistent_login_info)
{
    FUNCDEF("process_grid_files");
    // locate all grid definition files in the directory stated.
    directory grid_dir(grid_storage_folder, "*.def");

    if (_complain_on_missing_grid && !grid_dir.files().length()
        && (time_stamp().value() - _startup_time->value()
            > GRID_CHECK_PAUSE) ) {
      // there's no grid file after a substantial pause, that's worth
      // reporting to the user.
      LOG(isprintf("There are no grid definition files after %.1f minutes; "
          "this seems wrong.", float(GRID_CHECK_PAUSE) / MINUTE_ms));
      _complain_on_missing_grid = false;  // record that we already complained.
    }

    // for each file in that directory, process the database queries...
    for (int file_indy = 0; file_indy < grid_dir.files().length();
        file_indy++) {
      _complain_on_missing_grid = false;  // record that we saw a filename.

      istring current_file = grid_dir.files()[file_indy];
      if (_last_grid_seen != current_file) {
        LOG(istring("reading grid file: ") + current_file);
        _last_grid_seen = current_file;
      }
      // calculate which data source this grid file is for.
      istring just_ds = current_file.substring(0, current_file.length() - 5);
        // pare off the ".def" ending.
      int underscore_indy = just_ds.find('_', just_ds.end(), true);
      just_ds.zap(0, underscore_indy);
      int data_source = just_ds.convert(-1);
      if (data_source == -1) {
        LOG(istring("ERROR: failed to parse data source ID from file name: ")
            + current_file);
        continue;
      }
  //LOG(isprintf("ds id is %d", data_source));

      // create the full path to the grid definition.
      istring grid_def = grid_dir.path() + "/" + current_file;

      bool db_ready = _querier.prepare_for_query(grid_def, persistent_login_info);
      if (!db_ready) {
        // prepare_for_query already does enough logging.  just bail on this.
        reset_data_fields(df_storage_dir, data_source);
        continue;
      }

  //could be moved to a read db grid query method...
      // open grid definition as an INI file.
      ini_configurator ini(grid_def);
      // read the column names from the ini file.
      string_array col_names;
      float_array multipliers;  // values to multiply int fields by.
      istring curr_item = "bogus";  // satisfy loop condition.
      for (int i = 0; curr_item != ""; i++) {
        curr_item = ini.load(isprintf("column_%d", i), "dbcol", "");
        if (curr_item.t()) {
          istring mult = ini.load(isprintf("column_%d", i), "multiplier", "");
  //LOG(istring("found mult of ") + mult);
          if (mult.t()) multipliers += mult.convert(1.0);
          else multipliers += 1.0;
  //LOG(isprintf("calculated mult of %f", multipliers[multipliers.length() - 1]));
          col_names += curr_item;
        }
      }
      if (!col_names.length()) {
        LOG("ERROR: there was no column info garnered from grid definition.");
        reset_data_fields(df_storage_dir, data_source);
        continue;
      }

      istring key_column = ini.load("rows", "key_column", "");
      if (!key_column) {
        LOG("ERROR: the key column for row queries seems absent from grid definition.");
        reset_data_fields(df_storage_dir, data_source);
        continue;
      }

      // read the row information from the ini file.
      string_array row_keys;
      curr_item = "bogus";
      for (int i = 0; curr_item != ""; i++) {
        curr_item = ini.load("rows", isprintf("rowkey_%d", i), "");
        if (curr_item.t()) {
          row_keys += curr_item;
  //LOG(isprintf("row key #%d is ", i) + curr_item);
        }
      }
      if (!row_keys.length()) {
        LOG("ERROR: there was no row key info found in grid definition.");
        reset_data_fields(df_storage_dir, data_source);
        continue;
      }

      // first line in the query picks our escape character.
      istring query;
      // now we start the actual query for data.
      query += "select ";
      bool saw_key_col = false;
      for (int i = 0; i < col_names.length(); i++) {
        if (i) query += ", ";
        istring safe_colname = col_names[i];
        common_database_support::nerf_special_characters(safe_colname,
            DB_FREETDS_SQL_ESCAPE_CHARACTER);
        query += safe_colname;
        // make sure the query has the key column in it someplace.  this
        // allows us to know the order that the results should be shown in.
        if (col_names[i] == key_column) saw_key_col = true;
      }

      istring temp_key_col = key_column;
      common_database_support::nerf_special_characters(temp_key_col,
          DB_FREETDS_SQL_ESCAPE_CHARACTER);

      if (!saw_key_col) {
        // we assume there are other columns other than the key column; if not,
        // then what kind of query is this anyway?
        query += istring(", ") + temp_key_col;
      }

      query += " from ";
      query += persistent_login_info.table_name;
      query += " where ";
      query += temp_key_col;  // pre-nerfed.
      query += " in (";
      for (int i = 0; i < row_keys.length(); i++) {
        if (i) query += ", ";
        istring temp_rowkey = row_keys[i];
        common_database_support::nerf_special_characters(temp_rowkey,
            DB_FREETDS_SQL_ESCAPE_CHARACTER);
        // quotes are needed around string values and will work with non-strings.
        query += istring("'") + temp_rowkey + "'";
      }
      query += ")";
  //end of read db grid query method.

      // now chew on the database and store the results as data fields.
      bool worked = process_query_as_df(query, data_source, df_storage_dir,
          key_column, row_keys, multipliers);
      if (!worked) {
        LOG("ERROR: failed to process the query and acquire data fields.");
        reset_data_fields(df_storage_dir, data_source);
      }
      dbcanquery(_querier.current_db());
    }
  }

---