Artificial Intelligence Recommendation:Custom feature operators

Minimal example

Every custom operator follows the same pattern: inherit IFeatureOP, implement Initialize and at least one ProcessWith* method, and register with REGISTER_PLUGIN.

Here is a minimal operator that outputs a constant integer:

#include "api/base_op.h"

namespace fg {

class ConstantOp : public IFeatureOP {
 public:
  int Initialize(const string& feature_config) override {
    // Parse JSON config if needed; return 0 on success.
    return 0;
  }

  int ProcessWithInt32Outputs(const vector<FieldPtr>& inputs,
                              vector<int32>& outputs) override {
    outputs.push_back(42);
    return 0;  // 0 = success
  }
};

}  // namespace fg

REGISTER_PLUGIN("ConstantOp", ConstantOp);

Key requirements:

• Include a parameterless constructor (the default constructor works here).

• The Initialize method receives the entire fg.json entry as a JSON string. Parse configuration items you need from it.

• Return 0 for success from all methods; non-zero indicates failure.

• Call REGISTER_PLUGIN in the implementation file, not the header.

Choose between ProcessWith* and BatchProcess

Two processing interfaces are available. Choose based on your throughput and complexity requirements:

When BatchProcess is implemented and HasBatchProcessImpl() returns true, the framework skips all ProcessWith* methods.

Full C++ interface

#pragma once
#ifndef FEATURE_GENERATOR_PLUGIN_BASE_H
#define FEATURE_GENERATOR_PLUGIN_BASE_H

#include <absl/container/flat_hash_map.h>
#include <absl/strings/string_view.h>
#include <absl/types/optional.h>

#include <stdexcept>
#include <utility>
#include <vector>

#include "fsmap.h"
#include "integral_types.h"

namespace fg {

using absl::optional;
using std::string;
using std::vector;

template <typename T>
using List = std::vector<T>;
template <typename K, typename V>
using Map = absl::flat_hash_map<K, V>;
template <typename K, typename V>
using MapArray = std::vector<std::pair<K, V>>;
using Matrix = std::vector<std::vector<float>>;
using MatrixL = std::vector<std::vector<int64>>;
using MatrixS = std::vector<std::vector<string>>;
template <typename K, typename V>
using FSMap = featurestore::type::fs_map<K, V>;

using FieldPtr = absl::variant<
    const optional<string>*, const optional<int32>*, const optional<int64>*,
    const optional<float>*, const optional<double>*,
    const optional<absl::string_view>*,

    const List<string>*, const List<int32>*, const List<int64>*,
    const List<float>*, const List<double>*, const List<absl::string_view>*,

    const Map<string, string>*, const Map<string, int32>*,
    const Map<string, int64>*, const Map<string, float>*,
    const Map<string, double>*, const Map<string, absl::string_view>*,

    const Map<absl::string_view, absl::string_view>*,
    const Map<absl::string_view, int32>*, const Map<absl::string_view, int64>*,
    const Map<absl::string_view, float>*, const Map<absl::string_view, double>*,
    const Map<absl::string_view, string>*,

    const Map<int32, string>*, const Map<int32, int32>*,
    const Map<int32, int64>*, const Map<int32, float>*,
    const Map<int32, double>*, const Map<int32, absl::string_view>*,

    const Map<int64, string>*, const Map<int64, float>*,
    const Map<int64, double>*, const Map<int64, int32>*,
    const Map<int64, int64>*, const Map<int64, absl::string_view>*,

    const FSMap<absl::string_view, absl::string_view>*,
    const FSMap<absl::string_view, int32>*,
    const FSMap<absl::string_view, int64>*,
    const FSMap<absl::string_view, float>*,
    const FSMap<absl::string_view, double>*,

    const FSMap<int32, int32>*, const FSMap<int32, int64>*,
    const FSMap<int32, float>*, const FSMap<int32, double>*,
    const FSMap<int32, absl::string_view>*,

    const FSMap<int64, float>*, const FSMap<int64, double>*,
    const FSMap<int64, int32>*, const FSMap<int64, int64>*,
    const FSMap<int64, absl::string_view>*,

    const MapArray<string, string>*, const MapArray<string, int32>*,
    const MapArray<string, int64>*, const MapArray<string, float>*,
    const MapArray<string, double>*,

    const MapArray<int32, string>*, const MapArray<int32, float>*,
    const MapArray<int32, double>*, const MapArray<int32, int32>*,
    const MapArray<int32, int64>*,

    const MapArray<int64, string>*, const MapArray<int64, float>*,
    const MapArray<int64, double>*, const MapArray<int64, int32>*,
    const MapArray<int64, int64>*, const Matrix*, const MatrixL*,
    const MatrixS*>;

// Represents a COLUMN of the feature table.
using VariantVector = absl::variant<
    vector<optional<string>>, vector<optional<int32>>, vector<optional<int64>>,
    vector<optional<float>>, vector<optional<double>>,
    vector<optional<absl::string_view>>,

    vector<List<string>>, vector<List<int32>>, vector<List<int64>>,
    vector<List<float>>, vector<List<double>>, vector<List<absl::string_view>>,

    vector<Map<string, string>>, vector<Map<string, int32>>,
    vector<Map<string, int64>>, vector<Map<string, float>>,
    vector<Map<string, double>>, vector<Map<string, absl::string_view>>,

    vector<Map<absl::string_view, absl::string_view>>,
    vector<Map<absl::string_view, int32>>,
    vector<Map<absl::string_view, int64>>,
    vector<Map<absl::string_view, float>>,
    vector<Map<absl::string_view, double>>,

    vector<Map<int32, string>>, vector<Map<int32, int32>>,
    vector<Map<int32, int64>>, vector<Map<int32, float>>,
    vector<Map<int32, double>>, vector<Map<int32, absl::string_view>>,

    vector<Map<int64, string>>, vector<Map<int64, float>>,
    vector<Map<int64, double>>, vector<Map<int64, int32>>,
    vector<Map<int64, int64>>, vector<Map<int64, absl::string_view>>,

    vector<FSMap<absl::string_view, absl::string_view>>,
    vector<FSMap<absl::string_view, int32>>,
    vector<FSMap<absl::string_view, int64>>,
    vector<FSMap<absl::string_view, float>>,
    vector<FSMap<absl::string_view, double>>,

    vector<FSMap<int32, int32>>, vector<FSMap<int32, int64>>,
    vector<FSMap<int32, float>>, vector<FSMap<int32, double>>,
    vector<FSMap<int32, absl::string_view>>,

    vector<FSMap<int64, float>>, vector<FSMap<int64, double>>,
    vector<FSMap<int64, int32>>, vector<FSMap<int64, int64>>,
    vector<FSMap<int64, absl::string_view>>,

    vector<MapArray<string, string>>, vector<MapArray<string, int32>>,
    vector<MapArray<string, int64>>, vector<MapArray<string, float>>,
    vector<MapArray<string, double>>,

    vector<MapArray<int32, string>>, vector<MapArray<int32, float>>,
    vector<MapArray<int32, double>>, vector<MapArray<int32, int32>>,
    vector<MapArray<int32, int64>>,

    vector<MapArray<int64, string>>, vector<MapArray<int64, float>>,
    vector<MapArray<int64, double>>, vector<MapArray<int64, int32>>,
    vector<MapArray<int64, int64>>, vector<Matrix>, vector<MatrixL>,
    vector<MatrixS>>;

/**
 * @brief The public base class for custom feature operators.
 *
 * The framework checks if a subclass overrides the `BatchProcess` method. If it is overridden,
 * the framework calls this method to perform the feature transformation.
 * Otherwise, the framework selects one of the `ProcessWith*` methods based on the `value_type` configuration.
 * Implement the method that corresponds to the required output type.
 */
class IFeatureOP {
 public:
  class NotOverriddenException : public std::exception {
   public:
    explicit NotOverriddenException(std::string msg) : msg_(std::move(msg)) {}
    const char* what() const noexcept override {
      if (msg_.empty()) {
        return "unimplemented method called";
      }
      // Cache the message to a member variable to ensure that the returned pointer remains valid.
      cached_ = "unimplemented method called: " + msg_;
      return cached_.c_str();
    }

   private:
    std::string msg_;
    mutable std::string cached_;
  };

  virtual ~IFeatureOP() = default;

  /**
   * @brief Initialization method.
   * @param feature_config The full fg.json entry for this feature, as a JSON string.
   * @return 0 on success; non-zero indicates failure.
   */
  virtual int Initialize(const string& feature_config) = 0;

  /**
   * @brief Processes one record and outputs string values.
   * @param inputs A record that can contain multiple fields.
   * @param outputs The outputs of the feature transformation.
   * @return 0 on success.
   */
  virtual int ProcessWithStrOutputs(const vector<FieldPtr>& inputs,
                                    vector<string>& outputs) {
    throw NotOverriddenException("ProcessWithStrOutputs(FieldPtr)");
  }

  /**
   * @brief Processes one record and outputs int32 values.
   */
  virtual int ProcessWithInt32Outputs(const vector<FieldPtr>& inputs,
                                      vector<int32>& outputs) {
    throw NotOverriddenException("ProcessWithInt32Outputs(FieldPtr)");
  }

  /**
   * @brief Processes one record and outputs int64 values.
   */
  virtual int ProcessWithInt64Outputs(const vector<FieldPtr>& inputs,
                                      vector<int64>& outputs) {
    throw NotOverriddenException("ProcessWithInt64Outputs(FieldPtr)");
  }

  /**
   * @brief Processes one record and outputs float values.
   */
  virtual int ProcessWithFloatOutputs(const vector<FieldPtr>& inputs,
                                      vector<float>& outputs) {
    throw NotOverriddenException("ProcessWithFloatOutputs(FieldPtr)");
  }

  /**
   * @brief Processes one record and outputs double values.
   */
  virtual int ProcessWithDoubleOutputs(const vector<FieldPtr>& inputs,
                                       vector<double>& outputs) {
    throw NotOverriddenException("ProcessWithDoubleOutputs(FieldPtr)");
  }

  /**
   * @brief Optional batch interface that processes one batch of records.
   *
   * @param inputs A vector of input columns; each `VariantVector` represents one feature column.
   * @param outputs The transformed features. Supports complex output types usable as inputs for downstream operators.
   * @return 0 on success.
   */
  virtual int BatchProcess(const vector<VariantVector>& inputs,
                           VariantVector& outputs) {
    throw NotOverriddenException("BatchProcess");
  }

  /**
   * @brief Declares whether the subclass implements BatchProcess.
   *
   * Override this to return `true` if you implement `BatchProcess`.
   * This avoids exception propagation issues across dynamic library boundaries
   * when the .so and the main program use different C++ ABIs.
   *
   * @return true if BatchProcess is implemented; false otherwise (default).
   */
  virtual bool HasBatchProcessImpl() const { return false; }
};

using CreateOperatorFunc = IFeatureOP* (*)();

inline FieldPtr GetFieldPtr(const VariantVector& input, size_t i) {
  return absl::visit(
      [&](const auto& vec) -> FieldPtr {
        if (i >= vec.size()) {
          throw std::out_of_range("GetFieldPtr: index " + std::to_string(i) +
                                  " out of range [0, " +
                                  std::to_string(vec.size()) + ")");
        }
        return &vec.at(i);
      },
      input);
}
}  // namespace fg

#if defined(__GNUC__)
#define PLUGIN_API_HIDDEN \
  __attribute__((visibility("hidden"))) __attribute__((used))
#define PLUGIN_API_EXPORT \
  __attribute__((visibility("default"))) __attribute__((used))
#else
#define PLUGIN_API_HIDDEN
#define PLUGIN_API_EXPORT
#endif

std::vector<std::string>& getLocalNames();
std::vector<std::pair<std::string, void*>>& getLocalRegs();

#define REGISTER_PLUGIN(OpName, OpClass)                            \
  extern "C" PLUGIN_API_EXPORT fg::IFeatureOP* create##OpClass() {  \
    return new fg::OpClass();                                       \
  }                                                                 \
  namespace {                                                       \
  struct _Reg_##OpClass {                                           \
    _Reg_##OpClass() {                                              \
      getLocalNames().push_back(OpName);                            \
      getLocalRegs().emplace_back(OpName, (void*)&create##OpClass); \
    }                                                               \
  };                                                                \
  static _Reg_##OpClass _dummy_##OpClass __attribute__((used));     \
  }

#endif  // FEATURE_GENERATOR_PLUGIN_BASE_H

Implementation notes

• Input types: Implement ProcessWith* methods for all input types you intend to support. For unsupported types, throw an exception directly. VariantRecord defines all feature field types that the framework can process. FSMap types are required when integrating with Feature Store—they significantly improve online processor performance.

• Discretization: Implement only the transformation logic before discretization. If hash_bucket_size, vocab_list, boundaries, or another discretization operation is configured, the framework handles it automatically.

• Thread safety: By default (is_op_thread_safe=true), the operator must be stateless or use only thread_local variables. Set is_op_thread_safe=false to have the framework create one object replica per thread—this is simpler to implement but uses more memory.

• `string_view` inputs: Online services (EasyRecProcessor, TorchEasyRec Processor) pass item-side string features as absl::string_view for efficiency. If your operator can't handle string_view, set disable_string_view=true in the configuration to have the framework convert them to string before calling your operator. This degrades performance.

• Custom config items: Use any key names in the JSON entry—the framework passes the full JSON string to Initialize. Don't reuse framework-reserved key names (such as feature_type, operator_name, value_type). Keys that reference external files must end with _file so the framework can sync them for offline tasks.

• Operator directory: Use the FEATURE_OPERATOR_DIR environment variable to specify the directory where dynamic-link library files are located. Each dynamic-link library can contain implementations of multiple feature operators.

• `BatchProcess` return type: The type of the VariantVector returned by BatchProcess depends on the values of is_sequence, value_dimension, and value_type. For more information, see the output table schema in Configuration reference. When stub_type=true is configured and no binning operation is set, BatchProcess can return any valid type, such as Map.

fg.json entry structure

{
    "feature_name": "my_custom_fg_op",
    "feature_type": "custom_feature",
    "operator_name": "EditDistance",
    "operator_lib_file": "libedit_distance.so",
    "expression": [
        "user:query",
        "item:title"
    ],
    "value_type": "string",
    "separator": ",",
    "default_value": "-1",
    "value_dimension": 1,
    "normalizer": "method=expression,expr=x>16?16:x",
    "num_buckets": 10000,
    "stub_type": false,
    "is_sequence": false,
    "is_op_thread_safe": true
}

Add any additional configuration items your operator needs. The entire JSON entry is passed to Initialize.

Configuration parameters

Output table schema

The value_dimension and is_sequence settings determine the schema type of the output table in offline tasks:

Special cases: array<array<int>> is forced to array<array<bigint>>; array<array<double>> is forced to array<array<float>>.

Normalizer frameworks

For numeric features, add a normalizer to further process the transformation result:

For supported functions in expression, see Built-in feature operators.

Discretization operations

Six discretization types are available without any additional implementation:

For more information, see Feature discretization (binning).

Sequence time-difference feature

{
    "feature_name": "time_diff_seq",
    "feature_type": "custom_feature",
    "operator_name": "SeqExpr",
    "expression": ["user:cur_time", "user:clk_time_seq"],
    "formula": "cur_time - clk_time_seq",
    "default_value": "0",
    "value_type": "int32",
    "is_sequence": true,
    "num_buckets": 1000,
    "is_op_thread_safe": false
}

Spherical distance with normalization

{
    "feature_name": "spherical_distance",
    "feature_type": "custom_feature",
    "operator_name": "SeqExpr",
    "expression": ["item:click_id_lng", "item:click_id_lat", "user:j_lng", "user:j_lat"],
    "formula": "spherical_distance",
    "default_value": "0",
    "value_type": "double",
    "is_sequence": true,
    "is_op_thread_safe": true,
    "value_dimension": 1,
    "normalizer": "method=expression,expr=sqrt(x)"
}

Both examples use the SeqExpr operator. The formula field is a SeqExpr-specific configuration item passed through Initialize.

• spherical_distance: Calculates the distance between two latitude/longitude coordinate pairs. Parameters are [lng1_seq, lat1_seq, lng2, lat2]—the first two are sequences; the last two are scalars.

These examples demonstrate the tiled format for custom sequence features. For a nested-format example, see sequence_feature.

EditDistance configuration

For a BatchProcess example, download and review RegexReplace.