Method Make

cpp/src/arrow/acero/hash_join_node.cc:725–782

  }

  static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                const ExecNodeOptions& options) {
    // Number of input exec nodes must be 2
    RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 2, "HashJoinNode"));

    std::unique_ptr<HashJoinSchema> schema_mgr = std::make_unique<HashJoinSchema>();

    const auto& join_options = checked_cast<const HashJoinNodeOptions&>(options);
    RETURN_NOT_OK(ValidateHashJoinNodeOptions(join_options));

    const auto& left_schema = *(inputs[0]->output_schema());
    const auto& right_schema = *(inputs[1]->output_schema());

    // This will also validate input schemas
    if (join_options.output_all) {
      RETURN_NOT_OK(schema_mgr->Init(
          join_options.join_type, left_schema, join_options.left_keys, right_schema,
          join_options.right_keys, join_options.filter,
          join_options.output_suffix_for_left, join_options.output_suffix_for_right));
    } else {
      RETURN_NOT_OK(schema_mgr->Init(
          join_options.join_type, left_schema, join_options.left_keys,
          join_options.left_output, right_schema, join_options.right_keys,
          join_options.right_output, join_options.filter,
          join_options.output_suffix_for_left, join_options.output_suffix_for_right));
    }

    ARROW_ASSIGN_OR_RAISE(
        Expression filter,
        schema_mgr->BindFilter(join_options.filter, left_schema, right_schema,
                               plan->query_context()->exec_context()));

    // Generate output schema
    std::shared_ptr<Schema> output_schema = schema_mgr->MakeOutputSchema(
        join_options.output_suffix_for_left, join_options.output_suffix_for_right);

    // Create hash join implementation object
    // SwissJoin does not support:
    // a) 64-bit string offsets
    // b) dictionaries
    //
    bool use_swiss_join;
#if ARROW_LITTLE_ENDIAN
    use_swiss_join = !schema_mgr->HasDictionaries() && !schema_mgr->HasLargeBinary();
#else
    use_swiss_join = false;
#endif
    std::unique_ptr<HashJoinImpl> impl;
    if (use_swiss_join) {
      ARROW_ASSIGN_OR_RAISE(impl, HashJoinImpl::MakeSwiss());
    } else {
      ARROW_ASSIGN_OR_RAISE(impl, HashJoinImpl::MakeBasic());
    }

    return plan->EmplaceNode<HashJoinNode>(
        plan, inputs, join_options, std::move(output_schema), std::move(schema_mgr),
        std::move(filter), std::move(impl));
  }