extensions/sql/data/MaxCollector.h

/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <string>
#include <unordered_map>
#include <tuple>

#include "SQLRowSubscriber.h"

namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace sql {

class MaxCollector: public SQLRowSubscriber {
  void beginProcessRow() override {}

  void endProcessRow() override {
    if (columnsVerified_) {
      return;
    }

    if (countColumns_ != mapState_.size())
      throw minifi::Exception(PROCESSOR_EXCEPTION, "MaxCollector: Column(s) '" + maxValueColumnNames_ + "' are not found in the columns of '" + selectQuery_ + "' result.");

    columnsVerified_ = true;
  }

  void processColumnName(const std::string& name) override {
    if (columnsVerified_) {
      return;
    }

    if (mapState_.count(name)) {
      countColumns_++;
    }
  }

  void processColumn(const std::string& name, const std::string& value)  override {
    updateMaxValue(name, '\'' + value + '\'');
  }

  void processColumn(const std::string& name, double value) override {
    updateMaxValue(name, value);
  }

  void processColumn(const std::string& name, int value) override {
    updateMaxValue(name, value);
  }

  void processColumn(const std::string& name, long long value) override {
    updateMaxValue(name, value);
  }

  void processColumn(const std::string& name, unsigned long long value) override {
    updateMaxValue(name, value);
  }

  void processColumn(const std::string& name, const char* value) override {}

  template <typename T>
  struct MaxValue {
    void updateMaxValue(const std::string& name, const T& value) {
      const auto it = mapColumnNameValue_.find(name);
      if (it == mapColumnNameValue_.end()) {
        mapColumnNameValue_.insert({ name, value });
      } else {
        if (value > it->second) {
          it->second = value;
        }
      }
    }

    std::unordered_map<std::string, T> mapColumnNameValue_;
  };

  template <typename Tuple, int Index>
  struct UpdateMapState {
    UpdateMapState(const Tuple& tpl, std::unordered_map<std::string, std::string>& mapState) {
      for (auto& el : mapState) {
        const auto& maxVal = std::get<Index>(tpl);

        const auto it = maxVal.mapColumnNameValue_.find(el.first);
        if (it != maxVal.mapColumnNameValue_.end()) {
          std::stringstream ss;
          ss << it->second;
          el.second = ss.str();
        }
      }

      UpdateMapState<Tuple, Index - 1>(tpl, mapState);
    }
  };

  template <typename Tuple>
  struct UpdateMapState<Tuple, -1> {
    UpdateMapState(const Tuple&, std::unordered_map<std::string, std::string>&) {}
  };

  template <typename ...Ts>
  struct MaxValues : public std::tuple<MaxValue<Ts>...> {
    constexpr static size_t size = sizeof...(Ts);
  };

 public:
  MaxCollector(const std::string& selectQuery, const std::string& maxValueColumnNames, std::unordered_map<std::string, std::string>& mapState)
    :selectQuery_(selectQuery), maxValueColumnNames_(maxValueColumnNames), mapState_(mapState) {
  }

  template <typename T>
  void updateMaxValue(const std::string& columnName, const T& value) {
    if (mapState_.count(columnName)) {
      std::get<MaxValue<T>>(maxValues_).updateMaxValue(columnName, value);
    }
  }

  bool updateMapState() {
    auto mapState = mapState_;
    UpdateMapState<decltype(maxValues_), decltype(maxValues_)::size - 1>(maxValues_, mapState_);

    return mapState != mapState_;
  }

 private:
  const std::string selectQuery_;
  const std::string maxValueColumnNames_;
  std::unordered_map<std::string, std::string>& mapState_;
  MaxValues<std::string, double, int, long long, unsigned long long> maxValues_;
  size_t countColumns_{};
  bool columnsVerified_{false};
};
  
} /* namespace sql */
} /* namespace minifi */
} /* namespace nifi */
} /* namespace apache */
} /* namespace org */