[49/55] [abbrv] incubator-quickstep git commit: Initial commit

[ji...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/DateExtractOperation.cpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/DateExtractOperation.cpp b/types/operations/unary_operations/DateExtractOperation.cpp
index c99e403..c09b400 100644
--- a/types/operations/unary_operations/DateExtractOperation.cpp
+++ b/types/operations/unary_operations/DateExtractOperation.cpp
@@ -21,521 +21,136 @@
 
 #include <cstddef>
 #include <cstdint>
-#include <memory>
+#include <map>
 #include <string>
 #include <type_traits>
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-#include <utility>
 #include <vector>
 
-#include "storage/StorageBlockInfo.hpp"
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
-#include "catalog/CatalogTypedefs.hpp"
-#include "storage/ValueAccessor.hpp"
-#include "storage/ValueAccessorUtil.hpp"
+#include "types/DateType.hpp"
 #include "types/DatetimeLit.hpp"
-#include "types/IntType.hpp"
-#include "types/LongType.hpp"
+#include "types/DatetimeType.hpp"
 #include "types/Type.hpp"
-#include "types/TypeFactory.hpp"
 #include "types/TypeID.hpp"
 #include "types/TypedValue.hpp"
-#include "types/containers/ColumnVector.hpp"
-#include "types/operations/Operation.pb.h"
-#include "utility/Macros.hpp"
+#include "types/operations/unary_operations/UnaryOperationWrapper.hpp"
 
 #include "glog/logging.h"
 
-using std::int32_t;
-using std::int64_t;
-
 namespace quickstep {
 
-template <DateExtractUnit unit, bool argument_nullable>
-TypedValue DatetimeExtractUncheckedOperator<unit, argument_nullable>::applyToTypedValue(
-    const TypedValue &argument) const {
-  if (argument_nullable && argument.isNull()) {
-    return TypedValue(kLong);
-  }
-
-  return TypedValue(dateExtract(argument.getLiteral<DatetimeLit>()));
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-TypedValue DateExtractUncheckedOperator<unit, argument_nullable>::applyToTypedValue(
-    const TypedValue &argument) const {
-  if (argument_nullable && argument.isNull()) {
-    return TypedValue(kInt);
-  }
-
-  return TypedValue(dateExtract(argument.getLiteral<DateLit>()));
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-TypedValue DatetimeExtractUncheckedOperator<unit, argument_nullable>::applyToDataPtr(const void *argument) const {
-  if (argument_nullable && argument == nullptr) {
-    return TypedValue(kLong);
-  }
-
-  return TypedValue(dateExtract(*static_cast<const DatetimeLit*>(argument)));
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-TypedValue DateExtractUncheckedOperator<unit, argument_nullable>::applyToDataPtr(const void *argument) const {
-  if (argument_nullable && argument == nullptr) {
-    return TypedValue(kInt);
-  }
-
-  return TypedValue(dateExtract(*static_cast<const DateLit*>(argument)));
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DatetimeExtractUncheckedOperator<unit, argument_nullable>::applyToColumnVector(
-    const ColumnVector &argument) const {
-  // Datetime are usable with NativeColumnVector, so 'argument' should always
-  // be native.
-  DCHECK(argument.isNative());
-
-  const NativeColumnVector &native_argument = static_cast<const NativeColumnVector&>(argument);
-  std::unique_ptr<NativeColumnVector> result(
-      new NativeColumnVector(LongType::Instance(argument_nullable), native_argument.size()));
-
-  for (std::size_t pos = 0;
-       pos < native_argument.size();
-       ++pos) {
-    const DatetimeLit *datetime_arg =
-        static_cast<const DatetimeLit*>(native_argument.getUntypedValue<argument_nullable>(pos));
-    if (argument_nullable && (datetime_arg == nullptr)) {
-      result->appendNullValue();
-    } else {
-      *static_cast<int64_t*>(result->getPtrForDirectWrite())
-          = dateExtract(*datetime_arg);
-    }
-  }
-
-  return result.release();
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DateExtractUncheckedOperator<unit, argument_nullable>::applyToColumnVector(
-    const ColumnVector &argument) const {
-  // Date is usable with NativeColumnVector, so 'argument' should always
-  // be native.
-  DCHECK(argument.isNative());
-
-  const NativeColumnVector &native_argument = static_cast<const NativeColumnVector&>(argument);
-  std::unique_ptr<NativeColumnVector> result(
-      new NativeColumnVector(IntType::Instance(argument_nullable), native_argument.size()));
-
-  for (std::size_t pos = 0;
-       pos < native_argument.size();
-       ++pos) {
-    const DateLit *date_arg =
-        static_cast<const DateLit*>(native_argument.getUntypedValue<argument_nullable>(pos));
-    if (argument_nullable && (date_arg == nullptr)) {
-      result->appendNullValue();
-    } else {
-      *static_cast<int32_t*>(result->getPtrForDirectWrite())
-          = dateExtract(*date_arg);
-    }
-  }
-
-  return result.release();
-}
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DatetimeExtractUncheckedOperator<unit, argument_nullable>::applyToValueAccessor(
-    ValueAccessor *accessor,
-    const attribute_id argument_attr_id) const {
-  return InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
-      accessor,
-      [&](auto *accessor) -> ColumnVector* {  // NOLINT(build/c++11)
-    std::unique_ptr<NativeColumnVector> result(
-        new NativeColumnVector(LongType::Instance(argument_nullable), accessor->getNumTuples()));
-    accessor->beginIteration();
-    while (accessor->next()) {
-      const DatetimeLit *datetime_arg =
-          static_cast<const DatetimeLit*>(
-              accessor->template getUntypedValue<argument_nullable>(argument_attr_id));
-      if (argument_nullable && (datetime_arg == nullptr)) {
-        result->appendNullValue();
-      } else {
-        *static_cast<int64_t*>(result->getPtrForDirectWrite())
-            = this->dateExtract(*datetime_arg);
-      }
-    }
-    return result.release();
-  });
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DateExtractUncheckedOperator<unit, argument_nullable>::applyToValueAccessor(
-    ValueAccessor *accessor,
-    const attribute_id argument_attr_id) const {
-  return InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
-      accessor,
-      [&](auto *accessor) -> ColumnVector* {  // NOLINT(build/c++11)
-    std::unique_ptr<NativeColumnVector> result(
-        new NativeColumnVector(IntType::Instance(argument_nullable), accessor->getNumTuples()));
-    accessor->beginIteration();
-    while (accessor->next()) {
-      const DateLit *date_arg =
-          static_cast<const DateLit*>(
-              accessor->template getUntypedValue<argument_nullable>(argument_attr_id));
-      if (argument_nullable && (date_arg == nullptr)) {
-        result->appendNullValue();
-      } else {
-        *static_cast<int32_t*>(result->getPtrForDirectWrite())
-            = this->dateExtract(*date_arg);
-      }
-    }
-    return result.release();
-  });
-}
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DatetimeExtractUncheckedOperator<unit, argument_nullable>::applyToValueAccessorForJoin(
-    ValueAccessor *accessor,
-    const bool use_left_relation,
-    const attribute_id argument_attr_id,
-    const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const {
-  std::unique_ptr<NativeColumnVector> result(
-      new NativeColumnVector(LongType::Instance(argument_nullable), joined_tuple_ids.size()));
-  return InvokeOnValueAccessorNotAdapter(
-      accessor,
-      [&](auto *accessor) -> ColumnVector* {  // NOLINT(build/c++11)
-    for (const std::pair<tuple_id, tuple_id> &joined_pair : joined_tuple_ids) {
-      const DatetimeLit *datetime_arg =
-          static_cast<const DatetimeLit*>(
-              accessor->template getUntypedValueAtAbsolutePosition<argument_nullable>(
-                  argument_attr_id,
-                  use_left_relation ? joined_pair.first : joined_pair.second));
-      if (argument_nullable && (datetime_arg == nullptr)) {
-        result->appendNullValue();
-      } else {
-        *static_cast<int64_t*>(result->getPtrForDirectWrite())
-            = this->dateExtract(*datetime_arg);
-      }
-    }
-    return result.release();
-  });
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-ColumnVector* DateExtractUncheckedOperator<unit, argument_nullable>::applyToValueAccessorForJoin(
-    ValueAccessor *accessor,
-    const bool use_left_relation,
-    const attribute_id argument_attr_id,
-    const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const {
-  std::unique_ptr<NativeColumnVector> result(
-      new NativeColumnVector(IntType::Instance(argument_nullable), joined_tuple_ids.size()));
-  return InvokeOnValueAccessorNotAdapter(
-      accessor,
-      [&](auto *accessor) -> ColumnVector* {  // NOLINT(build/c++11)
-    for (const std::pair<tuple_id, tuple_id> &joined_pair : joined_tuple_ids) {
-      const DateLit *date_arg =
-          static_cast<const DateLit*>(
-              accessor->template getUntypedValueAtAbsolutePosition<argument_nullable>(
-                  argument_attr_id,
-                  use_left_relation ? joined_pair.first : joined_pair.second));
-      if (argument_nullable && (date_arg == nullptr)) {
-        result->appendNullValue();
-      } else {
-        *static_cast<int32_t*>(result->getPtrForDirectWrite())
-            = this->dateExtract(*date_arg);
-      }
-    }
-    return result.release();
-  });
-}
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
-template <DateExtractUnit unit, bool argument_nullable>
-inline int64_t
-DatetimeExtractUncheckedOperator<unit, argument_nullable>::dateExtract(
-    const DatetimeLit &argument) const {
-  switch (unit) {
-    case DateExtractUnit::kYear:
-      return argument.yearField();
-    case DateExtractUnit::kMonth:
-      return argument.monthField();
-    case DateExtractUnit::kDay:
-      return argument.dayField();
-    case DateExtractUnit::kHour:
-      return argument.hourField();
-    case DateExtractUnit::kMinute:
-      return argument.minuteField();
-    case DateExtractUnit::kSecond:
-      return argument.secondField();
-    default:
-      FATAL_ERROR("Unsupported DateExtractUnit in DatetimeExtractUncheckedOperator::dateExtract.");
-  }
-}
-
-template <DateExtractUnit unit, bool argument_nullable>
-inline int32_t DateExtractUncheckedOperator<unit, argument_nullable>::dateExtract(const DateLit &argument) const {
-  switch (unit) {
-    case DateExtractUnit::kYear:
-      return argument.yearField();
-    case DateExtractUnit::kMonth:
-      return argument.monthField();
-    default:
-      FATAL_ERROR("Unsupported DateExtractUnit in DateExtractUncheckedOperator::dateExtract.");
-  }
-}
-
-const DateExtractOperation& DateExtractOperation::Instance(const DateExtractUnit unit) {
-  switch (unit) {
-    case DateExtractUnit::kYear: {
-      static DateExtractOperation instance(DateExtractUnit::kYear);
-      return instance;
-    }
-    case DateExtractUnit::kMonth: {
-      static DateExtractOperation instance(DateExtractUnit::kMonth);
-      return instance;
-    }
-    case DateExtractUnit::kDay: {
-      static DateExtractOperation instance(DateExtractUnit::kDay);
-      return instance;
-    }
-    case DateExtractUnit::kHour: {
-      static DateExtractOperation instance(DateExtractUnit::kHour);
-      return instance;
-    }
-    case DateExtractUnit::kMinute: {
-      static DateExtractOperation instance(DateExtractUnit::kMinute);
-      return instance;
-    }
-    case DateExtractUnit::kSecond: {
-      static DateExtractOperation instance(DateExtractUnit::kSecond);
-      return instance;
-    }
-    default: {
-      FATAL_ERROR("Unsupported DateExtractUnit in DateExtractOperation::Instance.");
+struct DateExtractFunctor
+    : public UnaryFunctor<DateType, IntType> {
+  template <typename DateExtractUnitT>
+  inline int apply(const DateLit &argument) const {
+    switch (DateExtractUnitT::value) {
+      case DateExtractUnit::kYear:
+        return argument.yearField();
+      case DateExtractUnit::kMonth:
+        return argument.monthField();
+      case DateExtractUnit::kDay:
+        return argument.dayField();
+      default:
+        DLOG(FATAL) << "Unsupported DateExtractUnit in DateExtractFunctor::apply";
     }
   }
-}
-
-serialization::UnaryOperation DateExtractOperation::getProto() const {
-  serialization::UnaryOperation proto;
-  proto.set_operation_id(serialization::UnaryOperation::DATE_EXTRACT);
-
-  switch (unit_) {
-    case DateExtractUnit::kYear:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::YEAR);
-      break;
-    case DateExtractUnit::kMonth:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::MONTH);
-      break;
-    case DateExtractUnit::kDay:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::DAY);
-      break;
-    case DateExtractUnit::kHour:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::HOUR);
-      break;
-    case DateExtractUnit::kMinute:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::MINUTE);
-      break;
-    case DateExtractUnit::kSecond:
-      proto.SetExtension(serialization::DateExtractOperation::unit, serialization::DateExtractOperation::SECOND);
-      break;
-    default:
-      FATAL_ERROR("Unsupported DateExtractUnit in DateExtractOperation::getProto.");
-  }
-
-  return proto;
-}
-
-std::string DateExtractOperation::getName() const {
-  std::string name(kUnaryOperationNames[static_cast<std::size_t>(operation_id_)]);
-  name.push_back('(');
-  switch (unit_) {
-    case DateExtractUnit::kYear:
-      name.append("YEAR)");
-      break;
-    case DateExtractUnit::kMonth:
-      name.append("MONTH)");
-      break;
-    case DateExtractUnit::kDay:
-      name.append("DAY)");
-      break;
-    case DateExtractUnit::kHour:
-      name.append("HOUR)");
-      break;
-    case DateExtractUnit::kMinute:
-      name.append("MINUTE)");
-      break;
-    case DateExtractUnit::kSecond:
-      name.append("SECOND)");
-      break;
-    default:
-      name.append("UNKNOWN)");
-      break;
-  }
-  return name;
-}
-
-const Type* DateExtractOperation::pushDownTypeHint(const Type *type_hint) const {
-  if (type_hint == nullptr) {
-    return nullptr;
-  }
+};
 
-  if (type_hint->getTypeID() == kLong) {
-    switch (unit_) {
-      case DateExtractUnit::kYear:  // Fall through.
+struct DatetimeExtractFunctor
+    : public UnaryFunctor<DatetimeType, IntType> {
+  template <typename DateExtractUnitT>
+  inline std::int64_t apply(const DatetimeLit &argument) const {
+    switch (DateExtractUnitT::value) {
+      case DateExtractUnit::kYear:
+        return argument.yearField();
       case DateExtractUnit::kMonth:
-        // There are two possibilities for the return type, based on whether we
-        // have Datetime or Date as the underlying date implementation.
-        return nullptr;
-      case DateExtractUnit::kDay:  // Fall through.
+        return argument.monthField();
+      case DateExtractUnit::kDay:
+        return argument.dayField();
       case DateExtractUnit::kHour:
+        return argument.hourField();
       case DateExtractUnit::kMinute:
+        return argument.minuteField();
       case DateExtractUnit::kSecond:
-        return &TypeFactory::GetType(kDatetime, type_hint->isNullable());
+        return argument.secondField();
       default:
-        return nullptr;
+        DLOG(FATAL) << "Unsupported DateExtractUnit in DatetimeExtractFunctor::apply";
+    }
+  }
+};
+
+const std::map<std::string, DateExtractUnit> DateExtractOperation::kNameToUnitMap = {
+    { "year",   DateExtractUnit::kYear },
+    { "month",  DateExtractUnit::kMonth },
+    { "day",    DateExtractUnit::kDay },
+    { "hour",   DateExtractUnit::kHour },
+    { "minute", DateExtractUnit::kMinute },
+    { "second", DateExtractUnit::kSecond }
+};
+
+UncheckedUnaryOperator* DateExtractOperation::makeUncheckedUnaryOperator(
+    const Type &type,
+    const std::vector<TypedValue> &static_arguments) const {
+  DCHECK_EQ(1u, static_arguments.size());
+  DateExtractUnit unit = parseUnit(static_arguments.front());
+  const Type *result_type = getResultType(type, static_arguments);
+
+  if (type.getTypeID() == kDate) {
+    switch (unit) {
+      case DateExtractUnit::kYear:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DateExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kYear>>(
+                 type, *result_type);
+      case DateExtractUnit::kMonth:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DateExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kMonth>>(
+                 type, *result_type);
+      case DateExtractUnit::kDay:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DateExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kDay>>(
+                 type, *result_type);
+      default:
+        LOG(FATAL) << "Unsupported DateExtractUnit for DateType in "
+                   << "DateExtractOperation::makeUncheckedUnaryOperator";
     }
   } else {
-    return nullptr;
-  }
-}
-
-TypedValue DateExtractOperation::applyToChecked(const TypedValue &argument,
-                                                const Type &argument_type) const {
-  if (((argument.getTypeID() != TypeID::kDatetime) ||
-       (argument_type.getTypeID() != TypeID::kDatetime)) &&
-      ((argument.getTypeID() != TypeID::kDate) ||
-       (argument_type.getTypeID() != TypeID::kDate))) {
-    LOG(FATAL) << "UnaryOperation " << getName() << " is only applicable to Type "
-               << kTypeNames[TypeID::kDatetime] << ", but applyToChecked() was "
-               << "called with 'argument' of Type " << kTypeNames[argument.getTypeID()]
-               << " and explicit 'argument_type' parameter of "
-               << argument_type.getName();
-  }
-
-  if (argument.isNull()) {
-    if (argument.getTypeID() == TypeID::kDatetime) {
-      return TypedValue(kLong);
-    } else {
-      // argument type is kDate.
-      DCHECK_EQ(TypeID::kDate, argument.getTypeID());
-      return TypedValue(kInt);
-    }
-  }
-
-  switch (unit_) {
-    case DateExtractUnit::kYear: {
-      if (argument.getTypeID() == TypeID::kDatetime) {
-        return TypedValue(argument.getLiteral<DatetimeLit>().yearField());
-      } else {
-        // argument type is kDate.
-        DCHECK_EQ(TypeID::kDate, argument.getTypeID());
-        return TypedValue(argument.getLiteral<DateLit>().yearField());
-      }
-    }
-    case DateExtractUnit::kMonth: {
-      if (argument.getTypeID() == TypeID::kDatetime) {
-        return TypedValue(argument.getLiteral<DatetimeLit>().monthField());
-      } else {
-        // argument type is kDate.
-        DCHECK_EQ(TypeID::kDate, argument.getTypeID());
-        return TypedValue(argument.getLiteral<DateLit>().monthField());
-      }
-    }
-    case DateExtractUnit::kDay:
-      return TypedValue(argument.getLiteral<DatetimeLit>().dayField());
-    case DateExtractUnit::kHour:
-      return TypedValue(argument.getLiteral<DatetimeLit>().hourField());
-    case DateExtractUnit::kMinute:
-      return TypedValue(argument.getLiteral<DatetimeLit>().minuteField());
-    case DateExtractUnit::kSecond:
-      return TypedValue(argument.getLiteral<DatetimeLit>().secondField());
-    default: {
-      LOG(FATAL) << "Unsupported DateExtractUnit in DateExtractOperation::applyToChecked().";
-    }
-  }
-}
-
-UncheckedUnaryOperator* DateExtractOperation::makeUncheckedUnaryOperatorForTypeHelper(const Type &type) const {
-  switch (unit_) {
-    case DateExtractUnit::kYear: {
-      if (type.getTypeID() == TypeID::kDatetime) {
-        if (type.isNullable()) {
-          return new DatetimeExtractUncheckedOperator<DateExtractUnit::kYear, true>();
-        } else {
-          return new DatetimeExtractUncheckedOperator<DateExtractUnit::kYear, false>();
-        }
-      } else {
-        DCHECK_EQ(TypeID::kDate, type.getTypeID());
-        // type is kDate.
-        if (type.isNullable()) {
-          return new DateExtractUncheckedOperator<DateExtractUnit::kYear, true>();
-        } else {
-          return new DateExtractUncheckedOperator<DateExtractUnit::kYear, false>();
-        }
-      }
-    }
-    case DateExtractUnit::kMonth: {
-      if (type.getTypeID() == TypeID::kDatetime) {
-        if (type.isNullable()) {
-          return new DatetimeExtractUncheckedOperator<DateExtractUnit::kMonth, true>();
-        } else {
-          return new DatetimeExtractUncheckedOperator<DateExtractUnit::kMonth, false>();
-        }
-      } else {
-        // type is kDate.
-        DCHECK_EQ(TypeID::kDate, type.getTypeID());
-        if (type.isNullable()) {
-          return new DateExtractUncheckedOperator<DateExtractUnit::kMonth, true>();
-        } else {
-          return new DateExtractUncheckedOperator<DateExtractUnit::kMonth, false>();
-        }
-      }
-    }
-    case DateExtractUnit::kDay:
-      if (type.isNullable()) {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kDay, true>();
-      } else {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kDay, false>();
-      }
-    case DateExtractUnit::kHour:
-      if (type.isNullable()) {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kHour, true>();
-      } else {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kHour, false>();
-      }
-    case DateExtractUnit::kMinute:
-      if (type.isNullable()) {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kMinute, true>();
-      } else {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kMinute, false>();
-      }
-    case DateExtractUnit::kSecond:
-      if (type.isNullable()) {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kSecond, true>();
-      } else {
-        return new DatetimeExtractUncheckedOperator<DateExtractUnit::kSecond, false>();
-      }
-    default:
-      FATAL_ERROR("Unsupported DateExtractUnit in DateExtractOperation::makeUncheckedUnaryOperatorForTypeHelper.");
-  }
-}
-
-const Type* DateExtractOperation::resultTypeForArgumentType(const Type &type) const {
-  if (canApplyToType(type)) {
-    if (type.getTypeID() == kDatetime) {
-      return &LongType::Instance(type.isNullable());
-    } else {
-      DCHECK_EQ(kDate, type.getTypeID());
-      return &IntType::Instance(type.isNullable());
+    switch (unit) {
+      case DateExtractUnit::kYear:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kYear>>(
+                 type, *result_type);
+      case DateExtractUnit::kMonth:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kMonth>>(
+                 type, *result_type);
+      case DateExtractUnit::kDay:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kDay>>(
+                 type, *result_type);
+      case DateExtractUnit::kHour:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kHour>>(
+                 type, *result_type);
+      case DateExtractUnit::kMinute:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kMinute>>(
+                 type, *result_type);
+      case DateExtractUnit::kSecond:
+        return new UncheckedUnaryOperatorWrapperCodegen<
+             DatetimeExtractFunctor,
+             std::integral_constant<DateExtractUnit, DateExtractUnit::kSecond>>(
+                 type, *result_type);
+      default:
+        LOG(FATAL) << "Unsupported DateExtractUnit for DatetimeType in "
+                   << "DateExtractOperation::makeUncheckedUnaryOperator";
     }
-  } else {
-    return nullptr;
   }
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/DateExtractOperation.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/DateExtractOperation.hpp b/types/operations/unary_operations/DateExtractOperation.hpp
index f8c091b..00948a4 100644
--- a/types/operations/unary_operations/DateExtractOperation.hpp
+++ b/types/operations/unary_operations/DateExtractOperation.hpp
@@ -31,101 +31,26 @@
 #endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
 
 #include "catalog/CatalogTypedefs.hpp"
+#include "types/IntType.hpp"
 #include "types/LongType.hpp"
 #include "types/Type.hpp"
 #include "types/TypeID.hpp"
 #include "types/TypedValue.hpp"
 #include "types/operations/Operation.pb.h"
 #include "types/operations/unary_operations/UnaryOperation.hpp"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
 #include "utility/Macros.hpp"
+#include "utility/StringUtil.hpp"
 
 namespace quickstep {
 
-class ColumnVector;
-class ValueAccessor;
-
-struct DatetimeLit;
-
-/** \addtogroup Types
- *  @{
- */
-
 enum class DateExtractUnit {
   kYear = 0,
   kMonth,
   kDay,
   kHour,
   kMinute,
-  kSecond
-};
-
-/**
- * @brief UncheckedUnaryOperator for Datetime Extract.
- */
-template <DateExtractUnit unit, bool argument_nullable>
-class DatetimeExtractUncheckedOperator : public UncheckedUnaryOperator {
- public:
-  DatetimeExtractUncheckedOperator()
-      : UncheckedUnaryOperator() {}
-
-  TypedValue applyToTypedValue(const TypedValue &argument) const override;
-
-  TypedValue applyToDataPtr(const void *argument) const override;
-
-  ColumnVector* applyToColumnVector(const ColumnVector &argument) const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  ColumnVector* applyToValueAccessor(ValueAccessor *accessor,
-                                     const attribute_id argument_attr_id) const override;
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-  ColumnVector* applyToValueAccessorForJoin(
-      ValueAccessor *accessor,
-      const bool use_left_relation,
-      const attribute_id argument_attr_id,
-      const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const override;
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
- private:
-  inline std::int64_t dateExtract(const DatetimeLit &argument) const;
-
-  DISALLOW_COPY_AND_ASSIGN(DatetimeExtractUncheckedOperator);
-};
-
-/**
- * @brief UncheckedUnaryOperator for Date Extract.
- */
-template <DateExtractUnit unit, bool argument_nullable>
-class DateExtractUncheckedOperator : public UncheckedUnaryOperator {
- public:
-  DateExtractUncheckedOperator()
-      : UncheckedUnaryOperator() {}
-
-  TypedValue applyToTypedValue(const TypedValue &argument) const override;
-
-  TypedValue applyToDataPtr(const void *argument) const override;
-
-  ColumnVector* applyToColumnVector(const ColumnVector &argument) const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  ColumnVector* applyToValueAccessor(ValueAccessor *accessor,
-                                     const attribute_id argument_attr_id) const override;
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-  ColumnVector* applyToValueAccessorForJoin(
-      ValueAccessor *accessor,
-      const bool use_left_relation,
-      const attribute_id argument_attr_id,
-      const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const override;
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
- private:
-  inline std::int32_t dateExtract(const DateLit &argument) const;
-
-  DISALLOW_COPY_AND_ASSIGN(DateExtractUncheckedOperator);
+  kSecond,
+  kInvalid
 };
 
 /**
@@ -133,55 +58,67 @@ class DateExtractUncheckedOperator : public UncheckedUnaryOperator {
  */
 class DateExtractOperation : public UnaryOperation {
  public:
-  /**
-   * @brief Get a reference to the singleton instance of this Operation for a
-   *        particular DateExtractUnit.
-   *
-   * @param unit The date unit to extract.
-   * @return A reference to the singleton instance of this Operation for the
-   *         specified DateExtractUnit.
-   **/
-  static const DateExtractOperation& Instance(const DateExtractUnit unit);
-
-  serialization::UnaryOperation getProto() const override;
-
-  std::string getName() const override;
-
-  bool canApplyToType(const Type &type) const override {
-    return type.getTypeID() == TypeID::kDatetime || type.getTypeID() == kDate;
-  }
-
-  const Type* resultTypeForArgumentType(const Type &type) const override;
+  DateExtractOperation() {}
 
-  const Type* fixedNullableResultType() const override {
-    return nullptr;
+  std::string getName() const override {
+    return "DateExtract";
   }
 
-  bool resultTypeIsPlausible(const Type &result_type) const override {
-    return result_type.getTypeID() == kLong || result_type.getTypeID() == kInt;
+  std::string getShortName() const override {
+    return "DateExtract";
   }
 
-  const Type* pushDownTypeHint(const Type *type_hint) const override;
-
-  TypedValue applyToChecked(const TypedValue &argument,
-                            const Type &argument_type) const override;
+  std::vector<OperationSignaturePtr> getSignatures() const override {
+    const std::vector<TypeID> target_type_carrier = { kVarChar };
+    return {
+        OperationSignature::Create(getName(), {kDate}, target_type_carrier),
+        OperationSignature::Create(getName(), {kDatetime}, target_type_carrier)
+    };
+  }
 
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorForType(const Type &type) const override {
-    DCHECK(canApplyToType(type));
+  bool canApplyTo(const Type &type,
+                  const std::vector<TypedValue> &static_arguments,
+                  std::string *message) const override {
+    DCHECK_EQ(1u, static_arguments.size());
+    if (parseUnit(static_arguments.front()) == DateExtractUnit::kInvalid) {
+      *message = "Invalid extraction unit for DateExtract";
+      return false;
+    }
+    return true;
+  }
 
-    return makeUncheckedUnaryOperatorForTypeHelper(type);
+  const Type* getResultType(
+      const Type &type,
+      const std::vector<TypedValue> &static_arguments) const override {
+    DCHECK(UnaryOperation::canApplyTo(type, static_arguments));
+    if (type.getTypeID() == kDatetime) {
+      return &LongType::Instance(type.isNullable());
+    } else {
+      DCHECK_EQ(kDate, type.getTypeID());
+      return &IntType::Instance(type.isNullable());
+    }
   }
 
- private:
-  explicit DateExtractOperation(const DateExtractUnit unit)
-      : UnaryOperation(UnaryOperationID::kDateExtract),
-        unit_(unit) {}
+  UncheckedUnaryOperator* makeUncheckedUnaryOperator(
+      const Type &type,
+      const std::vector<TypedValue> &static_arguments) const override;
 
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorForTypeHelper(const Type &type) const;
+ private:
+  static DateExtractUnit parseUnit(const TypedValue &unit_arg) {
+    DCHECK(unit_arg.getTypeID() == kVarChar);
+    const std::string unit_str =
+        ToLower(std::string(static_cast<const char*>(unit_arg.getOutOfLineData())));
+
+    auto it = kNameToUnitMap.find(unit_str);
+    if (it != kNameToUnitMap.end()) {
+      return it->second;
+    } else {
+      return DateExtractUnit::kInvalid;
+    }
+  }
 
-  const DateExtractUnit unit_;
+  static const std::map<std::string, DateExtractUnit> kNameToUnitMap;
 
- private:
   DISALLOW_COPY_AND_ASSIGN(DateExtractOperation);
 };
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/NumericCastOperation.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/NumericCastOperation.hpp b/types/operations/unary_operations/NumericCastOperation.hpp
deleted file mode 100644
index 1c5e3d4..0000000
--- a/types/operations/unary_operations/NumericCastOperation.hpp
+++ /dev/null
@@ -1,313 +0,0 @@
-/**
- * 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.
- **/
-
-#ifndef QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_NUMERIC_CAST_OPERATION_HPP_
-#define QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_NUMERIC_CAST_OPERATION_HPP_
-
-#include <cstddef>
-#include <string>
-#include <utility>
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-#include <vector>
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
-#include "catalog/CatalogTypedefs.hpp"
-#include "storage/ValueAccessor.hpp"
-#include "storage/ValueAccessorUtil.hpp"
-#include "types/DoubleType.hpp"
-#include "types/FloatType.hpp"
-#include "types/IntType.hpp"
-#include "types/LongType.hpp"
-#include "types/Type.hpp"
-#include "types/TypeID.hpp"
-#include "types/TypedValue.hpp"
-#include "types/containers/ColumnVector.hpp"
-#include "types/operations/Operation.pb.h"
-#include "types/operations/unary_operations/UnaryOperation.hpp"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
-#include "utility/Macros.hpp"
-#include "utility/PtrMap.hpp"
-
-#include "glog/logging.h"
-
-namespace quickstep {
-
-/** \addtogroup Types
- *  @{
- */
-
-// TODO(quickstep-team): Support CAST on Datetime, YearMonthInterval, DatetimeInterval, VarChar, Char.
-//                       Currently we only need cast operations on numeric values.
-
-/**
- * @brief UncheckedUnaryOperator for CAST.
- */
-template <class SourceType, bool source_nullability, class TargetType, bool target_nullability>
-class UncheckedNumericCastOperator : public UncheckedUnaryOperator {
- public:
-  UncheckedNumericCastOperator()
-      : UncheckedUnaryOperator(),
-        target_type_(TargetType::Instance(target_nullability)) {
-  }
-
-  TypedValue applyToTypedValue(const TypedValue &argument) const override {
-    if (source_nullability && argument.isNull()) {
-      return TypedValue(TargetType::kStaticTypeID);
-    }
-
-    return TypedValue(static_cast<typename TargetType::cpptype>(
-        argument.getLiteral<typename SourceType::cpptype>()));
-  }
-
-  TypedValue applyToDataPtr(const void *argument) const override {
-    if (source_nullability && argument == nullptr) {
-      return TypedValue(TargetType::kStaticTypeID);
-    }
-
-    return TypedValue(
-        static_cast<const typename TargetType::cpptype>(
-            *static_cast<const typename SourceType::cpptype*>(argument)));
-  }
-
-  ColumnVector* applyToColumnVector(const ColumnVector &argument) const override {
-    DCHECK(NativeColumnVector::UsableForType(target_type_));
-    DCHECK(argument.isNative());
-    const NativeColumnVector &native_argument = static_cast<const NativeColumnVector&>(argument);
-    NativeColumnVector *result = new NativeColumnVector(
-        target_type_,
-        native_argument.size());
-    for (std::size_t pos = 0;
-         pos < native_argument.size();
-         ++pos) {
-      const typename SourceType::cpptype *scalar_arg
-          = static_cast<const typename SourceType::cpptype*>(
-              native_argument.getUntypedValue<source_nullability>(pos));
-      if (source_nullability && (scalar_arg == nullptr)) {
-        result->appendNullValue();
-      } else {
-        *static_cast<typename TargetType::cpptype*>(result->getPtrForDirectWrite())
-            = static_cast<typename TargetType::cpptype>(*scalar_arg);
-      }
-    }
-    return result;
-  }
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  ColumnVector* applyToValueAccessor(ValueAccessor *accessor,
-                                     const attribute_id argument_attr_id) const override {
-    DCHECK(NativeColumnVector::UsableForType(target_type_));
-    return InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
-        accessor,
-        [&](auto *accessor) -> ColumnVector* {  // NOLINT(build/c++11)
-      NativeColumnVector *result = new NativeColumnVector(
-          target_type_,
-          accessor->getNumTuples());
-      accessor->beginIteration();
-      while (accessor->next()) {
-        const typename SourceType::cpptype *scalar_arg
-            = static_cast<const typename SourceType::cpptype*>(
-                accessor->template getUntypedValue<source_nullability>(argument_attr_id));
-        if (source_nullability && (scalar_arg == nullptr)) {
-          result->appendNullValue();
-        } else {
-          *static_cast<typename TargetType::cpptype*>(result->getPtrForDirectWrite())
-              = static_cast<typename TargetType::cpptype>(*scalar_arg);
-        }
-      }
-      return result;
-    });
-  }
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-  ColumnVector* applyToValueAccessorForJoin(
-      ValueAccessor *accessor,
-      const bool use_left_relation,
-      const attribute_id argument_attr_id,
-      const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const override {
-    DCHECK(NativeColumnVector::UsableForType(target_type_));
-    NativeColumnVector *result = new NativeColumnVector(target_type_,
-                                                        joined_tuple_ids.size());
-    InvokeOnValueAccessorNotAdapter(
-        accessor,
-        [&](auto *accessor) -> void {  // NOLINT(build/c++11)
-      for (const std::pair<tuple_id, tuple_id> &joined_pair : joined_tuple_ids) {
-        const typename SourceType::cpptype *scalar_arg
-            = static_cast<const typename SourceType::cpptype*>(
-                accessor->template getUntypedValueAtAbsolutePosition<source_nullability>(
-                    argument_attr_id,
-                    use_left_relation ? joined_pair.first : joined_pair.second));
-        if (source_nullability && (scalar_arg == nullptr)) {
-          result->appendNullValue();
-        } else {
-          *static_cast<typename TargetType::cpptype*>(result->getPtrForDirectWrite())
-              = static_cast<typename TargetType::cpptype>(*scalar_arg);
-        }
-      }
-    });
-    return result;
-  }
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
- private:
-  const Type &target_type_;
-
-  DISALLOW_COPY_AND_ASSIGN(UncheckedNumericCastOperator);
-};
-
-/**
- * @brief UnaryOperation for CAST.
- */
-class NumericCastOperation : public UnaryOperation {
- public:
-  serialization::UnaryOperation getProto() const override {
-    serialization::UnaryOperation proto;
-    proto.set_operation_id(serialization::UnaryOperation::CAST);
-    proto.MutableExtension(serialization::CastOperation::target_type)
-        ->CopyFrom(getTargetType().getProto());
-
-    return proto;
-  }
-
-  /**
-   * @brief Get a reference to the singleton instance of this Operation.
-   *
-   * @param target_type The target type to coerce input values to.
-   * @return A reference to the singleton instance of this Operation.
-   **/
-  static const NumericCastOperation& Instance(const Type &target_type) {
-    static PtrMap<const Type*, NumericCastOperation> instance_map;
-    PtrMap<const Type*, NumericCastOperation>::iterator instance_map_it =
-        instance_map.find(&target_type);
-    if (instance_map_it == instance_map.end()) {
-      instance_map_it = instance_map.insert(&target_type,
-                                            new NumericCastOperation(target_type)).first;
-    }
-    return *(instance_map_it->second);
-  }
-
-  /**
-   * @return The target type for coercion.
-   */
-  const Type& getTargetType() const {
-    return target_type_;
-  }
-
-  std::string getName() const override {
-    return std::string(kUnaryOperationNames[static_cast<std::size_t>(operation_id_)])
-        .append("(")
-        .append(target_type_.getName())
-        .append(")");
-  }
-
-  bool canApplyToType(const Type &type) const override {
-    return target_type_.isCoercibleFrom(type);
-  }
-
-  const Type* resultTypeForArgumentType(const Type &type) const override {
-    if (canApplyToType(type)) {
-      return &target_type_;
-    } else {
-      return nullptr;
-    }
-  }
-
-  const Type* fixedNullableResultType() const override {
-    return &target_type_.getNullableVersion();
-  }
-
-  bool resultTypeIsPlausible(const Type &result_type) const override {
-    return result_type.equals(target_type_)
-           || result_type.equals(target_type_.getNullableVersion());
-  }
-
-  const Type* pushDownTypeHint(const Type *type_hint) const override {
-    return &target_type_;
-  }
-
-  TypedValue applyToChecked(const TypedValue &argument,
-                            const Type &argument_type) const override {
-    return target_type_.coerceValue(argument, argument_type);
-  }
-
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorForType(const Type &type) const override {
-    switch (type.getTypeID()) {
-      case kInt:
-        return makeUncheckedUnaryOperatorHelperForSourceNullability<IntType>(type);
-      case kLong:
-        return makeUncheckedUnaryOperatorHelperForSourceNullability<LongType>(type);
-      case kFloat:
-        return makeUncheckedUnaryOperatorHelperForSourceNullability<FloatType>(type);
-      case kDouble:
-        return makeUncheckedUnaryOperatorHelperForSourceNullability<DoubleType>(type);
-      default:
-        FATAL_ERROR("Unhandled type " << kTypeNames[type.getTypeID()]);
-    }
-  }
-
- private:
-  explicit NumericCastOperation(const Type &target_type)
-      : UnaryOperation(UnaryOperationID::kCast),
-        target_type_(target_type) {}
-
-  template <class SourceType>
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorHelperForSourceNullability(const Type &type) const {
-    if (type.isNullable()) {
-      return makeUncheckedUnaryOperatorHelperForTargetType<SourceType, true>();
-    } else {
-      return makeUncheckedUnaryOperatorHelperForTargetType<SourceType, false>();
-    }
-  }
-
-  template <class SourceType, bool source_nullability>
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorHelperForTargetType() const {
-    switch (target_type_.getTypeID()) {
-      case kInt:
-        return makeUncheckedUnaryOperatorHelperForTargetNullability<SourceType, source_nullability, IntType>();
-      case kLong:
-        return makeUncheckedUnaryOperatorHelperForTargetNullability<SourceType, source_nullability, LongType>();
-      case kFloat:
-        return makeUncheckedUnaryOperatorHelperForTargetNullability<SourceType, source_nullability, FloatType>();
-      case kDouble:
-        return makeUncheckedUnaryOperatorHelperForTargetNullability<SourceType, source_nullability, DoubleType>();
-      default:
-        FATAL_ERROR("Unhandled type " << kTypeNames[target_type_.getTypeID()]);
-    }
-  }
-
-  template <class SourceType, bool source_nullability, class TargetType>
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorHelperForTargetNullability() const {
-    if (target_type_.isNullable()) {
-      return new UncheckedNumericCastOperator<SourceType, source_nullability, TargetType, true>();
-    } else {
-      return new UncheckedNumericCastOperator<SourceType, source_nullability, TargetType, false>();
-    }
-  }
-
-  const Type& target_type_;
-
-  DISALLOW_COPY_AND_ASSIGN(NumericCastOperation);
-};
-
-/** @} */
-
-}  // namespace quickstep
-
-#endif  // QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_NUMERIC_CAST_OPERATION_HPP_

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/SubstringOperation.cpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/SubstringOperation.cpp b/types/operations/unary_operations/SubstringOperation.cpp
index 84f1c8d..408ea2d 100644
--- a/types/operations/unary_operations/SubstringOperation.cpp
+++ b/types/operations/unary_operations/SubstringOperation.cpp
@@ -40,30 +40,27 @@
 
 namespace quickstep {
 
-serialization::UnaryOperation SubstringOperation::getProto() const {
-  serialization::UnaryOperation proto;
-  proto.set_operation_id(serialization::UnaryOperation::SUBSTRING);
-  proto.SetExtension(serialization::SubstringOperation::start_position,
-                     start_position_);
-  proto.SetExtension(serialization::SubstringOperation::substring_length,
-                     substring_length_);
-  return proto;
-}
+UncheckedUnaryOperator* SubstringOperation::makeUncheckedUnaryOperator(
+    const Type &type,
+    const std::vector<TypedValue> &static_arguments) const {
+  DCHECK(UnaryOperation::canApplyTo(type, static_arguments));
 
-UncheckedUnaryOperator* SubstringOperation::makeUncheckedUnaryOperatorForType(
-    const Type &type) const {
-  DCHECK(type.getSuperTypeID() == Type::kAsciiString);
+  std::size_t start_position;
+  std::size_t substring_length;
+  ExtractStaticArguments(static_arguments, &start_position, &substring_length);
 
   const std::size_t input_maximum_length =
-      static_cast<const AsciiStringSuperType&>(type).getStringLength();
+      type.getTypeID() == kChar
+          ? static_cast<const CharType&>(type).getStringLength()
+          : static_cast<const VarCharType&>(type).getStringLength();
   const bool input_null_terminated = (type.getTypeID() == TypeID::kVarChar);
 
-  const Type *result_type = resultTypeForArgumentType(type);
+  const Type *result_type = getResultType(type, static_arguments);
   DCHECK(result_type != nullptr);
 
   return CreateBoolInstantiatedInstance<SubstringUncheckedOperator, UncheckedUnaryOperator>(
-      std::forward_as_tuple(start_position_,
-                            computeMaximumSubstringLength(type),
+      std::forward_as_tuple(start_position,
+                            ComputeMaximumSubstringLength(type, start_position, substring_length),
                             input_maximum_length,
                             *result_type),
       input_null_terminated, type.isNullable());
@@ -108,23 +105,6 @@ TypedValue SubstringUncheckedOperator<null_terminated,
 }
 
 template <bool null_terminated, bool input_nullable>
-TypedValue SubstringUncheckedOperator<null_terminated,
-                                      input_nullable>
-    ::applyToDataPtr(const void *argument) const {
-  if (input_nullable && argument == nullptr) {
-    return TypedValue(result_type_.getTypeID());
-  }
-
-  char *output_ptr = static_cast<char*>(std::malloc(substring_length_));
-  computeSubstring(static_cast<const char*>(argument),
-                   output_ptr);
-
-  return TypedValue::CreateWithOwnedData(result_type_.getTypeID(),
-                                         output_ptr,
-                                         substring_length_);
-}
-
-template <bool null_terminated, bool input_nullable>
 ColumnVector* SubstringUncheckedOperator<null_terminated,
                                          input_nullable>
     ::applyToColumnVector(const ColumnVector &argument) const {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/SubstringOperation.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/SubstringOperation.hpp b/types/operations/unary_operations/SubstringOperation.hpp
index 66f311f..fd5868f 100644
--- a/types/operations/unary_operations/SubstringOperation.hpp
+++ b/types/operations/unary_operations/SubstringOperation.hpp
@@ -22,23 +22,27 @@
 
 #include <algorithm>
 #include <cstddef>
+#include <cstdint>
 #include <cstdlib>
 #include <cstring>
+#include <limits>
 #include <memory>
 #include <unordered_map>
 #include <utility>
 #include <vector>
 
 #include "catalog/CatalogTypedefs.hpp"
+#include "types/CharType.hpp"
 #include "types/Type.hpp"
 #include "types/TypeFactory.hpp"
 #include "types/TypeID.hpp"
 #include "types/TypedValue.hpp"
+#include "types/VarCharType.hpp"
 #include "types/operations/Operation.pb.h"
 #include "types/operations/unary_operations/UnaryOperation.hpp"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
 #include "utility/HashPair.hpp"
 #include "utility/Macros.hpp"
+#include "utility/StringUtil.hpp"
 
 #include "glog/logging.h"
 
@@ -53,116 +57,81 @@ class ValueAccessor;
  */
 class SubstringOperation : public UnaryOperation {
  public:
-  /**
-   * @brief Get a reference to the singleton instance of this Operation for
-   *        the given (start_position, substring_length) pair.
-   **/
-  static const SubstringOperation& Instance(const std::size_t start_position,
-                                            const std::size_t substring_length) {
-    // TODO(jianqiao): This is a temporary solution that creates a new instance
-    // for each distinct pair of start_position and substring_length arguments.
-    // The number of instances may be unbounded if quickstep continuously accepts
-    // queries that call SUBSTRING with different arguments. It still remains to
-    // design a better long-term solution.
-    const auto hash = [](const auto &pair) {
-      return hash_combine_detail::HashCombiner<std::size_t>::CombineHashes(pair.first, pair.second);
-    };
-    static std::unordered_map<std::pair<std::size_t, std::size_t>,
-                              std::unique_ptr<const SubstringOperation>,
-                              decltype(hash)> instance_map(10, hash);
-
-    const std::pair<std::size_t, std::size_t> key_pair =
-        std::make_pair(start_position, substring_length);
-    auto imit = instance_map.find(key_pair);
-    if (imit != instance_map.end()) {
-      return *imit->second;
-    } else {
-      const SubstringOperation *instance =
-          new SubstringOperation(start_position, substring_length);
-      instance_map.emplace(key_pair,
-                           std::unique_ptr<const SubstringOperation>(instance));
-      return *instance;
-    }
+  SubstringOperation() {}
+
+  std::string getName() const override {
+    return "Substring";
   }
 
-  serialization::UnaryOperation getProto() const override;
+  std::string getShortName() const override {
+    return "Substring";
+  }
 
-  bool canApplyToType(const Type &type) const override {
-    return (type.getSuperTypeID() == Type::kAsciiString);
+  std::vector<OperationSignaturePtr> getSignatures() const override {
+    return {
+        OperationSignature::Create(getName(), {kChar}, {kLong, kLong}),
+        OperationSignature::Create(getName(), {kVarChar}, {kLong, kLong})
+    };
   }
 
-  const Type *resultTypeForArgumentType(const Type &type) const override {
-    if (type.getSuperTypeID() == Type::kAsciiString) {
-      // Result is a Char string.
-      return &TypeFactory::GetType(TypeID::kChar,
-                                   computeMaximumSubstringLength(type),
-                                   type.isNullable());
+  bool canApplyTo(const Type &type,
+                  const std::vector<TypedValue> &static_arguments,
+                  std::string *message) const override {
+    DCHECK(type.getTypeID() == kChar || type.getTypeID() == kVarChar);
+    DCHECK(!static_arguments.empty() && static_arguments[0].getTypeID() == kLong);
+    DCHECK(static_arguments.size() <= 2);
+
+    if (static_arguments[0].getLiteral<std::int64_t>() <= 0) {
+      *message = "The start position must be greater than 0";
+      return false;
     }
-    return nullptr;
-  }
 
-  const Type* fixedNullableResultType() const override {
-    // Result type is not fixed (i.e. can have various lengths).
-    return nullptr;
-  }
+    if (static_arguments.size() == 2) {
+      DCHECK(static_arguments[1].getTypeID() == kLong);
+      if (static_arguments[1].getLiteral<std::int64_t>() <= 0) {
+        *message = "The substring length must be greater than 0";
+        return false;
+      }
+    }
 
-  bool resultTypeIsPlausible(const Type &result_type) const override {
-    // Result can be coerced to Char or VarChar.
-    return (result_type.getSuperTypeID() == Type::kAsciiString);
+    return true;
   }
 
-  const Type* pushDownTypeHint(const Type *type_hint) const override {
-    // Input can only be a string, but we don't know the length.
-    return nullptr;
-  }
+  const Type* getResultType(
+      const Type &type,
+      const std::vector<TypedValue> &static_arguments) const override {
+    DCHECK(UnaryOperation::canApplyTo(type, static_arguments));
 
-  TypedValue applyToChecked(const TypedValue &argument,
-                            const Type &argument_type) const override {
-    DCHECK(canApplyToType(argument_type));
-
-    const Type *result_type = resultTypeForArgumentType(argument_type);
-    DCHECK(result_type != nullptr);
-
-    if (argument_type.isNullable() && argument.isNull()) {
-      return result_type->makeNullValue();
-    } else {
-      const std::size_t result_length = computeMaximumSubstringLength(argument_type);
-      char *output_ptr = static_cast<char*>(std::malloc(result_length));
-      const char *input_ptr = static_cast<const char*>(argument.getOutOfLineData());
-
-      const std::size_t string_length = argument.getAsciiStringLength();
-      if (start_position_ >= string_length) {
-        *output_ptr = '\0';
-      } else {
-        const std::size_t actual_substring_length =
-            std::min(string_length - start_position_, substring_length_);
-        std::memcpy(output_ptr, input_ptr + start_position_, actual_substring_length);
-        if (actual_substring_length < result_length) {
-          output_ptr[actual_substring_length] = '\0';
-        }
-      }
+    std::size_t start_position;
+    std::size_t substring_length;
+    ExtractStaticArguments(static_arguments, &start_position, &substring_length);
 
-      return TypedValue::CreateWithOwnedData(result_type->getTypeID(),
-                                             output_ptr,
-                                             result_length);
-    }
+    return &TypeFactory::GetType(TypeID::kChar,
+                                 ComputeMaximumSubstringLength(type, start_position, substring_length),
+                                 type.isNullable());
   }
 
-  UncheckedUnaryOperator* makeUncheckedUnaryOperatorForType(const Type &type) const override;
+  UncheckedUnaryOperator* makeUncheckedUnaryOperator(
+      const Type &type,
+      const std::vector<TypedValue> &static_arguments) const override;
 
  private:
-  /**
-   * @brief Constructor.
-   *
-   * @param input_type The data type of the input argument for substring.
-   * @param start_position The 0-base starting position of the substring.
-   * @param substring_length The substring length.
-   */
-  SubstringOperation(const std::size_t start_position,
-                     const std::size_t substring_length)
-      : UnaryOperation(UnaryOperationID::kSubstring),
-        start_position_(start_position),
-        substring_length_(substring_length) {
+  inline static void ExtractStaticArguments(
+      const std::vector<TypedValue> &static_arguments,
+      std::size_t *start_position,
+      std::size_t *substring_length) {
+    DCHECK_LE(1u, static_arguments.size());
+    DCHECK_GE(2u, static_arguments.size());
+
+    DCHECK(static_arguments[0].getTypeID() == kLong);
+    *start_position =
+        static_cast<std::size_t>(static_arguments[0].getLiteral<std::int64_t>() - 1);
+
+    DCHECK(static_arguments.size() < 2u || static_arguments[1].getTypeID() == kLong);
+    *substring_length =
+        static_arguments.size() < 2u
+            ? std::numeric_limits<std::size_t>::max()
+            : static_cast<std::size_t>(static_arguments[1].getLiteral<std::int64_t>());
   }
 
   /**
@@ -171,19 +140,23 @@ class SubstringOperation : public UnaryOperation {
    *
    * @param type The type of the input, must be either CharType or VarCharType.
    */
-  inline std::size_t computeMaximumSubstringLength(const Type& type) const {
-      DCHECK(type.getSuperTypeID() == Type::kAsciiString);
-
-      // Substring result should have length no greater than the minimum of
-      // (1) the input string length subtract the start position, and
-      // (2) the specified substring length.
-     return std::min(static_cast<const AsciiStringSuperType&>(type).getStringLength() - start_position_,
-                     substring_length_);
+  inline static std::size_t ComputeMaximumSubstringLength(
+      const Type& type,
+      const std::size_t start_position,
+      const std::size_t substring_length) {
+    DCHECK(type.getTypeID() == kChar || type.getTypeID() == kVarChar);
+
+    const std::size_t input_maximum_length =
+        type.getTypeID() == kChar
+            ? static_cast<const CharType&>(type).getStringLength()
+            : static_cast<const VarCharType&>(type).getStringLength();
+
+    // Substring result should have length no greater than the minimum of
+    // (1) the input string length subtract the start position, and
+    // (2) the specified substring length.
+    return std::min(input_maximum_length - start_position, substring_length);
   }
 
-  const std::size_t start_position_;
-  const std::size_t substring_length_;
-
  private:
   DISALLOW_COPY_AND_ASSIGN(SubstringOperation);
 };
@@ -203,8 +176,6 @@ class SubstringUncheckedOperator : public UncheckedUnaryOperator {
 
   TypedValue applyToTypedValue(const TypedValue& argument) const override;
 
-  TypedValue applyToDataPtr(const void *argument) const override;
-
   ColumnVector* applyToColumnVector(const ColumnVector &argument) const override;
 
 #ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperation.cpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperation.cpp b/types/operations/unary_operations/UnaryOperation.cpp
index af150b3..09b27d9 100644
--- a/types/operations/unary_operations/UnaryOperation.cpp
+++ b/types/operations/unary_operations/UnaryOperation.cpp
@@ -20,28 +20,8 @@
 #include "types/operations/unary_operations/UnaryOperation.hpp"
 
 #include "types/operations/Operation.pb.h"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
 #include "utility/Macros.hpp"
 
 namespace quickstep {
 
-serialization::UnaryOperation UnaryOperation::getProto() const {
-  serialization::UnaryOperation proto;
-  switch (operation_id_) {
-    case UnaryOperationID::kNegate:
-      proto.set_operation_id(serialization::UnaryOperation::NEGATE);
-      break;
-    case UnaryOperationID::kCast:
-      FATAL_ERROR("Must use the overridden NumericCastOperation::getProto");
-    case UnaryOperationID::kDateExtract:
-      FATAL_ERROR("Must use the overridden DateExtractOperation::getProto");
-    case UnaryOperationID::kSubstring:
-      FATAL_ERROR("Must use the overridden SubstringOperation::getProto");
-    default:
-      FATAL_ERROR("Unrecognized UnaryOperationID in UnaryOperation::getProto");
-  }
-
-  return proto;
-}
-
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperation.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperation.hpp b/types/operations/unary_operations/UnaryOperation.hpp
index 30a2961..68dc993 100644
--- a/types/operations/unary_operations/UnaryOperation.hpp
+++ b/types/operations/unary_operations/UnaryOperation.hpp
@@ -24,18 +24,10 @@
 #include <string>
 #include <type_traits>
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-#include <utility>
-#include <vector>
-
-#include "storage/StorageBlockInfo.hpp"
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
 #include "catalog/CatalogTypedefs.hpp"
-#include "types/operations/Operation.hpp"
-#include "types/operations/Operation.pb.h"
 #include "types/TypedValue.hpp"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
+#include "types/operations/Operation.hpp"
+#include "types/operations/OperationSignature.hpp"
 #include "utility/Macros.hpp"
 
 namespace quickstep {
@@ -48,6 +40,9 @@ class ValueAccessor;
  *  @{
  */
 
+class UnaryOperation;
+typedef std::shared_ptr<const UnaryOperation> UnaryOperationPtr;
+
 /**
  * @brief A unary operator which can be quickly applied to data items WITHOUT
  *        checking their type.
@@ -69,14 +64,6 @@ class UncheckedUnaryOperator {
   virtual TypedValue applyToTypedValue(const TypedValue &argument) const = 0;
 
   /**
-   * @brief Apply to a data item via a pointer without type-checking.
-   *
-   * @param argument The data item to apply to.
-   * @return The literal result of the operation.
-   **/
-  virtual TypedValue applyToDataPtr(const void *argument) const = 0;
-
-  /**
    * @brief Apply to a vector of values without type-checking.
    *
    * @param argument The argument ColumnVector to apply to.
@@ -96,27 +83,6 @@ class UncheckedUnaryOperator {
                                              const attribute_id argument_attr_id) const = 0;
 #endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-  /**
-   * @brief Apply to an attribute of a list of joined tuples in a
-   *        ValueAccessor.
-   *
-   * @param accessor The ValueAccessor to apply to.
-   * @param use_left_relation If true, this UnaryOperation's argument is
-   *        assumed to be taken from the left relation in the pairs of
-   *        joined_tuple_ids. If false, the right relation.
-   * @param argument_attr_id The attribute ID of the argument in accessor.
-   * @param joined_tuple_ids A series of pairs of tuple ids from the left and
-   *        right relations in a join.
-   * @return A ColumnVector of literal results of the operation.
-   **/
-  virtual ColumnVector* applyToValueAccessorForJoin(
-      ValueAccessor *accessor,
-      const bool use_left_relation,
-      const attribute_id argument_attr_id,
-      const std::vector<std::pair<tuple_id, tuple_id>> &joined_tuple_ids) const = 0;
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_JOIN
-
  protected:
   UncheckedUnaryOperator() {
   }
@@ -134,30 +100,11 @@ class UncheckedUnaryOperator {
 class UnaryOperation : public Operation {
  public:
   /**
-   * @brief Generate a serialized Protocol Buffer representation of
-   *        this UnaryOperation.
-   *
-   * @return The serialized Protocol Buffer representation of this UnaryOperation.
-   **/
-  virtual serialization::UnaryOperation getProto() const;
-
-  /**
-   * @brief Determine the ID of this UnaryOperation.
-   *
-   * @return The ID of this UnaryOperation.
-   **/
-  inline UnaryOperationID getUnaryOperationID() const {
-    return operation_id_;
-  }
-
-  /**
    * @brief Get a human-readable name for this UnaryOperation.
    *
    * @return A human-readable name for this UnaryOperation.
    **/
-  virtual std::string getName() const {
-    return kUnaryOperationNames[static_cast<std::size_t>(operation_id_)];
-  }
+  virtual std::string getName() const = 0;
 
   /**
    * @brief Get a human-readable short name (e.g. "-") for this UnaryOperation.
@@ -165,113 +112,35 @@ class UnaryOperation : public Operation {
    * @return A human-readable short name for this BinaryOperation.
    **/
   virtual std::string getShortName() const {
-    return kUnaryOperationShortNames[static_cast<std::size_t>(operation_id_)];
+    return getName();
   }
 
-  /**
-   * @brief Determine whether this UnaryOperation can apply to the specified
-   *        Type.
-   *
-   * @param type The argument Type to check.
-   * @return Whether this UnaryOperation can apply to type.
-   **/
-  virtual bool canApplyToType(const Type &type) const = 0;
+  virtual std::vector<OperationSignaturePtr> getSignatures() const = 0;
 
-  /**
-   * @brief Determine the Type of the result from applying this UnaryOperation
-   *        to an argument of the specified Type.
-   *
-   * @param type The argument Type to check.
-   * @return The Type of the result from applying this UnaryOperation to type
-   *         (NULL if not applicable).
-   **/
-  virtual const Type* resultTypeForArgumentType(const Type &type) const = 0;
+  virtual bool canApplyTo(const Type &argument_type,
+                          const std::vector<TypedValue> &static_arguments,
+                          std::string *message) const = 0;
 
-  /**
-   * @brief If this UnaryOperation always yields the same Type (or if the ONLY
-   *        difference between 2 possible return Types is nullability), return
-   *        that Type, otherwise return NULL.
-   *
-   * @return The nullable version of this UnaryOperation's fixed result Type,
-   *         if applicable.
-   **/
-  virtual const Type* fixedNullableResultType() const = 0;
+  virtual const Type* getResultType(
+      const Type &argument_type,
+      const std::vector<TypedValue> &static_arguments) const = 0;
 
-  /**
-   * @brief Check if a particular Type might possibly be returned by this
-   *        UnaryOperation, assuming an appropriate argument type.
-   * @note A nullable result type may be considered plausible even if a
-   *       particular UnaryOperation never actually returns NULL values, so
-   *       long as the non-nullable version of the type would otherwise be
-   *       plausible.
-   *
-   * @param result_type Check whether this Type can possibly be returned by
-   *        this UnaryOperation.
-   * @return true if result_type can be returned by this UnaryOperation, false
-   *         otherwise.
-   **/
-  virtual bool resultTypeIsPlausible(const Type &result_type) const = 0;
+  virtual UncheckedUnaryOperator* makeUncheckedUnaryOperator(
+      const Type &argument_type,
+      const std::vector<TypedValue> &static_arguments) const = 0;
 
-  /**
-   * @brief Get a "hint" Type for the argument to this UnaryOperation based on
-   *        a hint for this UnaryOperation's result type. If possible, returns
-   *        a pointer to a Type that, when given to this UnaryOperation as an
-   *        argument, yields values of the desired type (i.e. calling
-   *        resultTypeForArgumentType() on the returned type should return the
-   *        original type_hint).
-   * @note In some cases (e.g. NumericCastOperation) there may be multiple
-   *       types that can be used as arguments to this UnaryOperation that will
-   *       all yield the desired type_hint. In such cases, this method will
-   *       pick one Type based on its own implementation-specific preference.
-   *
-   * @param type_hint A hint about what Type the result of this UnaryOperation
-   *        should have. May be NULL to indicate no preference.
-   * @return A type hint for the argument to this UnaryOperation based on
-   *         type_hint, or NULL if no suitable Type exists.
-   **/
-  virtual const Type* pushDownTypeHint(const Type *type_hint) const = 0;
-
-  /**
-   * @brief Apply this UnaryOperation to a TypedValue.
-   * @warning It is an error to call this method if this UnaryOperation can not
-   *          be applied to argument_type. If in doubt, check canApplyToType()
-   *          first.
-   *
-   * @param argument The TypedValue to apply to.
-   * @param argument_type The Type that argument belongs to.
-   * @return The literal result of the operation.
-   **/
-  virtual TypedValue applyToChecked(const TypedValue &argument,
-                                    const Type &argument_type) const = 0;
-
-  /**
-   * @brief Create an UncheckedUnaryOperator which can apply to items of the
-   *        specified type.
-   * @warning The resulting UncheckedUnaryOperator performs no type-checking
-   *          whatsoever. Nonetheless, it is useful in situations where many
-   *          data items of the same, known type are to be operated on (for
-   *          example, over many tuples in the same relation).
-   *
-   * @param type The Type of argument to apply to.
-   * @return An UncheckedUnaryOperator which applies this UnaryOperation to
-   *         the specified Type.
-   * @exception OperationInapplicableToType This UnaryOperation is not
-   *            applicable to type.
-   **/
-  virtual UncheckedUnaryOperator* makeUncheckedUnaryOperatorForType(const Type &type) const = 0;
+  bool canApplyTo(
+      const Type &argument_type,
+      const std::vector<TypedValue> &static_arguments) const {
+    std::string message;
+    return canApplyTo(argument_type, static_arguments, &message);
+  }
 
  protected:
-  explicit UnaryOperation(const UnaryOperationID operation_id)
-      : Operation(Operation::kUnaryOperation,
-                  kUnaryOperationNames[
-                      static_cast<typename std::underlying_type<UnaryOperationID>::type>(operation_id)],
-                  kUnaryOperationShortNames[
-                      static_cast<typename std::underlying_type<UnaryOperationID>::type>(operation_id)]),
-        operation_id_(operation_id) {
+  UnaryOperation()
+      : Operation(Operation::kUnaryOperation) {
   }
 
-  const UnaryOperationID operation_id_;
-
  private:
   DISALLOW_COPY_AND_ASSIGN(UnaryOperation);
 };

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperationFactory.cpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperationFactory.cpp b/types/operations/unary_operations/UnaryOperationFactory.cpp
index b306061..cebf9e2 100644
--- a/types/operations/unary_operations/UnaryOperationFactory.cpp
+++ b/types/operations/unary_operations/UnaryOperationFactory.cpp
@@ -21,100 +21,54 @@
 
 #include <string>
 
-#include "types/TypeFactory.hpp"
-#include "types/operations/Operation.pb.h"
+#include "types/operations/OperationSignature.hpp"
 #include "types/operations/unary_operations/ArithmeticUnaryOperations.hpp"
-#include "types/operations/unary_operations/NumericCastOperation.hpp"
+#include "types/operations/unary_operations/CMathUnaryOperations.hpp"
+#include "types/operations/unary_operations/CastOperation.hpp"
 #include "types/operations/unary_operations/DateExtractOperation.hpp"
 #include "types/operations/unary_operations/SubstringOperation.hpp"
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
+#include "types/operations/unary_operations/UnaryOperationWrapper.hpp"
 #include "utility/Macros.hpp"
-
-#include "glog/logging.h"
+#include "utility/StringUtil.hpp"
 
 namespace quickstep {
 
-const UnaryOperation& UnaryOperationFactory::GetUnaryOperation(const UnaryOperationID id) {
-  switch (id) {
-    case UnaryOperationID::kNegate:
-      return NegateUnaryOperation::Instance();
-    case UnaryOperationID::kCast:
-      FATAL_ERROR("Getting a CastOperation through GetUnaryOperation is not supported");
-    case UnaryOperationID::kDateExtract:
-      FATAL_ERROR("Getting a DateExtractOperation through GetUnaryOperation is not supported");
-    case UnaryOperationID::kSubstring:
-      FATAL_ERROR("Getting a SubstringOperation through GetUnaryOperation is not supported");
-    default:
-      FATAL_ERROR("Unknown UnaryOperationID");
-  }
+UnaryOperationFactory::UnaryOperationFactory() {
+  registerUnaryOperation(UnaryOperationPtr(new CastOperation()));
+  registerUnaryOperation(UnaryOperationPtr(new DateExtractOperation()));
+  registerUnaryOperation(UnaryOperationPtr(new SubstringOperation()));
+
+  registerUnaryOperationPack<CMathUnaryFunctorPack>();
+  registerUnaryOperationPack<ArithmeticUnaryFunctorPack>();
 }
 
-bool UnaryOperationFactory::ProtoIsValid(const serialization::UnaryOperation &proto) {
-  // Check that UnaryOperation is fully initialized.
-  if (!proto.IsInitialized()) {
-    return false;
-  }
+void UnaryOperationFactory::registerUnaryOperation(
+    const UnaryOperationPtr &operation) {
+  for (const OperationSignaturePtr op_sig : operation->getSignatures()) {
+    const OperationSignaturePtr normalized_op_sig =
+        OperationSignature::Create(ToLower(op_sig->getName()),
+                                   op_sig->getArgumentTypeIDs(),
+                                   op_sig->getNumStaticArguments());
 
-  // Check that the operation_id is a valid UnaryOperation.
-  if (!proto.UnaryOperationID_IsValid(proto.operation_id())) {
-    return false;
-  }
+    // TODO: print error message for collision
+    unary_operations_.emplace(normalized_op_sig, operation);
 
-  switch (proto.operation_id()) {
-    case serialization::UnaryOperation::NEGATE:
-      return true;
-    case serialization::UnaryOperation::CAST:
-      return proto.HasExtension(serialization::CastOperation::target_type)
-          && TypeFactory::ProtoIsValid(proto.GetExtension(serialization::CastOperation::target_type));
-    case serialization::UnaryOperation::DATE_EXTRACT:
-      return proto.HasExtension(serialization::DateExtractOperation::unit)
-          && DateExtractOperation_Unit_IsValid(proto.GetExtension(serialization::DateExtractOperation::unit));
-    case serialization::UnaryOperation::SUBSTRING:
-      return proto.HasExtension(serialization::SubstringOperation::start_position)
-          && proto.HasExtension(serialization::SubstringOperation::substring_length);
-    default:
-      return false;
+    const auto name_arity_pair = std::make_pair(normalized_op_sig->getName(),
+                                                normalized_op_sig->getArity());
+    name_arity_index_[name_arity_pair].emplace(normalized_op_sig);
   }
 }
 
-const UnaryOperation& UnaryOperationFactory::ReconstructFromProto(
-    const serialization::UnaryOperation &proto) {
-  DCHECK(ProtoIsValid(proto))
-      << "Attempted to create UnaryOperation from an invalid proto description:\n"
-      << proto.DebugString();
-
-  switch (proto.operation_id()) {
-    case serialization::UnaryOperation::NEGATE:
-      return GetUnaryOperation(UnaryOperationID::kNegate);
-    case serialization::UnaryOperation::CAST:
-      return NumericCastOperation::Instance(
-          TypeFactory::ReconstructFromProto(
-              proto.GetExtension(
-                  serialization::CastOperation::target_type)));
-    case serialization::UnaryOperation::DATE_EXTRACT:
-      switch (proto.GetExtension(serialization::DateExtractOperation::unit)) {
-        case serialization::DateExtractOperation::YEAR:
-          return DateExtractOperation::Instance(DateExtractUnit::kYear);
-        case serialization::DateExtractOperation::MONTH:
-          return DateExtractOperation::Instance(DateExtractUnit::kMonth);
-        case serialization::DateExtractOperation::DAY:
-          return DateExtractOperation::Instance(DateExtractUnit::kDay);
-        case serialization::DateExtractOperation::HOUR:
-          return DateExtractOperation::Instance(DateExtractUnit::kHour);
-        case serialization::DateExtractOperation::MINUTE:
-          return DateExtractOperation::Instance(DateExtractUnit::kMinute);
-        case serialization::DateExtractOperation::SECOND:
-          return DateExtractOperation::Instance(DateExtractUnit::kSecond);
-        default:
-          FATAL_ERROR("Unrecognized DateExtractOperation unit in UnaryOperation::ReconstructFromProto");
-      }
-    case serialization::UnaryOperation::SUBSTRING:
-      return SubstringOperation::Instance(
-          proto.GetExtension(serialization::SubstringOperation::start_position),
-          proto.GetExtension(serialization::SubstringOperation::substring_length));
-    default:
-      FATAL_ERROR("Unrecognized UnaryOperationID in UnaryOperation::ReconstructFromProto");
+template <typename UnaryOperationPackT>
+void UnaryOperationFactory::registerUnaryOperationPack() {
+  for (const UnaryOperationPtr &operation : UnaryOperationPackT::GenerateAll()) {
+    registerUnaryOperation(operation);
   }
 }
 
+const UnaryOperationFactory& UnaryOperationFactory::Instance() {
+  static UnaryOperationFactory instance;
+  return instance;
+}
+
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperationFactory.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperationFactory.hpp b/types/operations/unary_operations/UnaryOperationFactory.hpp
index 2ce83d4..c586b62 100644
--- a/types/operations/unary_operations/UnaryOperationFactory.hpp
+++ b/types/operations/unary_operations/UnaryOperationFactory.hpp
@@ -20,13 +20,18 @@
 #ifndef QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_UNARY_OPERATION_FACTORY_HPP_
 #define QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_UNARY_OPERATION_FACTORY_HPP_
 
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
+#include <set>
+#include <unordered_map>
+#include <utility>
+
+#include "types/operations/OperationSignature.hpp"
+#include "types/operations/unary_operations/UnaryOperation.hpp"
+#include "utility/HashPair.hpp"
 #include "utility/Macros.hpp"
 
-namespace quickstep {
+#include "glog/logging.h"
 
-class UnaryOperation;
-namespace serialization { class UnaryOperation; }
+namespace quickstep {
 
 /** \addtogroup Types
  *  @{
@@ -37,38 +42,51 @@ namespace serialization { class UnaryOperation; }
  **/
 class UnaryOperationFactory {
  public:
-  /**
-   * @brief Convenience factory method to get a pointer to a UnaryOperation
-   *        from that UnaryOperation's ID.
-   *
-   * @param id The ID of the desired UnaryOperation.
-   * @return The UnaryOperation corresponding to id.
-   **/
-  static const UnaryOperation& GetUnaryOperation(const UnaryOperationID id);
-
-  /**
-   * @brief Get a reference to a UnaryOperation from that UnaryOperation's
-   *        serialized Protocol Buffer representation.
-   *
-   * @param proto A serialized Protocol Buffer representation of a UnaryOperation,
-   *        originally generated by getProto().
-   * @return The UnaryOperation described by proto.
-   **/
-  static const UnaryOperation& ReconstructFromProto(const serialization::UnaryOperation &proto);
-
-  /**
-   * @brief Check whether a serialization::UnaryOperation is fully-formed and
-   *        all parts are valid.
-   *
-   * @param proto A serialized Protocol Buffer representation of a UnaryOperation,
-   *        originally generated by getProto().
-   * @return Whether proto is fully-formed and valid.
-   **/
-  static bool ProtoIsValid(const serialization::UnaryOperation &proto);
+  static const UnaryOperationFactory& Instance();
+
+  inline bool hasUnaryOperation(
+      const OperationSignaturePtr &op_signature) const {
+    return unary_operations_.find(op_signature) != unary_operations_.end();
+  }
+
+  inline bool hasUnaryOperation(const std::string &name,
+                                const std::size_t arity) const {
+    const auto it = name_arity_index_.find(std::make_pair(name, arity));
+    return it != name_arity_index_.end();
+  }
+
+  inline const UnaryOperationPtr getUnaryOperation(
+      const OperationSignaturePtr &op_signature) const {
+    DCHECK(hasUnaryOperation(op_signature));
+    return unary_operations_.at(op_signature);
+  }
+
+  inline const std::set<OperationSignaturePtr,
+                        OperationSignatureNumStaticArgumentsGreater>&
+      getUnaryOperations(const std::string &name,
+                         const std::size_t arity) const {
+    DCHECK(hasUnaryOperation(name, arity));
+    return name_arity_index_.at(std::make_pair(name, arity));
+  }
 
  private:
   UnaryOperationFactory();
 
+  void registerUnaryOperation(const UnaryOperationPtr &operation);
+
+  template <typename UnaryOperationPackT>
+  void registerUnaryOperationPack();
+
+  std::unordered_map<OperationSignaturePtr,
+                     UnaryOperationPtr,
+                     OperationSignatureHash,
+                     OperationSignatureEqual> unary_operations_;
+
+  std::unordered_map<
+      std::pair<std::string, std::size_t>,
+      std::set<OperationSignaturePtr,
+               OperationSignatureNumStaticArgumentsGreater>> name_arity_index_;
+
   DISALLOW_COPY_AND_ASSIGN(UnaryOperationFactory);
 };
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperationID.cpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperationID.cpp b/types/operations/unary_operations/UnaryOperationID.cpp
deleted file mode 100644
index b47a848..0000000
--- a/types/operations/unary_operations/UnaryOperationID.cpp
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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.
- **/
-
-#include "types/operations/unary_operations/UnaryOperationID.hpp"
-
-namespace quickstep {
-
-const char *kUnaryOperationNames[] = {
-  "Negate", "Cast", "DateExtract", "Substring"
-};
-
-const char *kUnaryOperationShortNames[] = {
-  "-", "Cast", "DateExtract", "Substring"
-};
-
-}  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/6941f903/types/operations/unary_operations/UnaryOperationID.hpp
----------------------------------------------------------------------
diff --git a/types/operations/unary_operations/UnaryOperationID.hpp b/types/operations/unary_operations/UnaryOperationID.hpp
deleted file mode 100644
index fa50f50..0000000
--- a/types/operations/unary_operations/UnaryOperationID.hpp
+++ /dev/null
@@ -1,63 +0,0 @@
-/**
- * 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.
- **/
-
-#ifndef QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_UNARY_OPERATION_ID_HPP_
-#define QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_UNARY_OPERATION_ID_HPP_
-
-#include <type_traits>
-
-namespace quickstep {
-
-/** \addtogroup Types
- *  @{
- */
-
-/**
- * @brief Concrete UnaryOperations.
- **/
-enum class UnaryOperationID {
-  kNegate = 0,
-  kCast,
-  kDateExtract,
-  kSubstring,
-  kNumUnaryOperationIDs  // Not a real UnaryOperationID, exists for counting purposes.
-};
-
-/**
- * @brief Names of comparisons in the same order as UnaryOperationID.
- * @note Defined out-of-line in UnaryOperation.cpp
- **/
-extern const char *kUnaryOperationNames[
-    static_cast<typename std::underlying_type<UnaryOperationID>::type>(
-        UnaryOperationID::kNumUnaryOperationIDs)];
-
-/**
- * @brief Short names (i.e. mathematical symbols) of comparisons in the same
- *        order as UnaryOperationID.
- * @note Defined out-of-line in UnaryOperation.cpp
- **/
-extern const char *kUnaryOperationShortNames[
-    static_cast<typename std::underlying_type<UnaryOperationID>::type>(
-        UnaryOperationID::kNumUnaryOperationIDs)];
-
-/** @} */
-
-}  // namespace quickstep
-
-#endif  // QUICKSTEP_TYPES_OPERATIONS_UNARY_OPERATIONS_UNARY_OPERATION_ID_HPP_