Method initializeFilter

bolt/exec/MergeJoin.cpp:128–198

}

void MergeJoin::initializeFilter(
    const core::TypedExprPtr& filter,
    const RowTypePtr& leftType,
    const RowTypePtr& rightType) {
  std::vector<core::TypedExprPtr> filters = {filter};
  filter_ =
      std::make_unique<ExprSet>(std::move(filters), operatorCtx_->execCtx());

  column_index_t filterChannel = 0;
  std::vector<std::string> names;
  std::vector<TypePtr> types;
  auto numFields = filter_->expr(0)->distinctFields().size();
  names.reserve(numFields);
  types.reserve(numFields);

  // This function is called in a loop for all columns used in a filter
  // expression. For each column, 'channel' specifies column index in the left
  // or right side of the join input, 'outputs' specifies the mapping from input
  // columns of that side of the join to the output of the join; 'filters'
  // specifies the mapping from input columns to the 'filterInput_' for columns
  // that are not projected to the output.
  //
  // This function checks whether the column is projected to the output and if
  // so adds an entry to 'filterInputToOutputChannel_'. Otherwise, adds an entry
  // to 'filters'.
  //
  // At the end of the loop, each column used in a filter appears either in:
  //
  //  - filterInputToOutputChannel_: in case the column is projected to the join
  //  output;
  //  - filterLeftInputs_: in case the column is not projected to the join
  //  output and it comes from the left side of the join;
  //  - filterRightInputs_: in case the column is not projected to the join
  //  output and it comes from the right side of the join.
  auto addChannel = [&](column_index_t channel,
                        const std::vector<IdentityProjection>& outputs,
                        std::vector<IdentityProjection>& filters,
                        const RowTypePtr& inputType) {
    names.emplace_back(inputType->nameOf(channel));
    types.emplace_back(inputType->childAt(channel));

    for (const auto& [inputChannel, outputChannel] : outputs) {
      if (inputChannel == channel) {
        filterInputToOutputChannel_.emplace(filterChannel++, outputChannel);
        return;
      }
    }
    filters.emplace_back(channel, filterChannel++);
  };

  for (const auto& field : filter_->expr(0)->distinctFields()) {
    const auto& name = field->field();
    auto channel = leftType->getChildIdxIfExists(name);
    if (channel.has_value()) {
      addChannel(
          channel.value(), leftProjections_, filterLeftInputs_, leftType);
      continue;
    }