// 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 "exec/join-builder.h"

#include "service/hs2-util.h"
#include "util/debug-util.h"
#include "util/min-max-filter.h"
#include "util/runtime-profile-counters.h"

#include "common/names.h"

namespace impala {

Status JoinBuilderConfig::Init(
    const TDataSink& tsink, const RowDescriptor* input_row_desc, FragmentState* state) {
  RETURN_IF_ERROR(DataSinkConfig::Init(tsink, input_row_desc, state));
  join_node_id_ = tsink.join_build_sink.dest_node_id;
  join_op_ = tsink.join_build_sink.join_op;
  return Status::OK();
}

JoinBuilder::JoinBuilder(TDataSinkId sink_id, const JoinBuilderConfig& sink_config,
    const string& name, RuntimeState* state)
  : DataSink(sink_id, sink_config, name, state),
    join_node_id_(sink_config.join_node_id_),
    join_op_(sink_config.join_op_),
    is_separate_build_(sink_id != -1),
    num_probe_threads_(
        is_separate_build_ ? state->instance_ctx().num_join_build_outputs : 1),
    outstanding_probes_(num_probe_threads_) {
  DCHECK_GE(outstanding_probes_, 1);
}

JoinBuilder::~JoinBuilder() {
  DCHECK_EQ(0, probe_refcount_);
}

void JoinBuilder::CloseFromProbe(RuntimeState* join_node_state) {
  if (is_separate_build_) {
    bool last_probe;
    {
      unique_lock<mutex> l(separate_build_lock_);
      --probe_refcount_;
      last_probe = probe_refcount_ == 0;
      VLOG(3) << "JoinBuilder (id=" << join_node_id_ << ")"
              << "closed from probe from finstance "
              << PrintId(join_node_state->fragment_instance_id())
              << "probe_refcount_=" << probe_refcount_;
      DCHECK_GE(probe_refcount_, 0);
    }
    // Only need to notify when the probe count is zero.
    if (last_probe) build_wakeup_cv_.NotifyAll();
  } else {
    Close(join_node_state);
  }
}

void JoinBuilder::CloseBeforeProbe(RuntimeState* join_node_state) {
  if (is_separate_build_) {
    bool no_threads_remaining;
    {
      unique_lock<mutex> l(separate_build_lock_);
      --outstanding_probes_;
      // If all the probes are done (probe_refcount_ == 0) and there are no
      // futher probes expected (outstanding_probes_ == 0), then there are
      // no threads left.
      no_threads_remaining = (probe_refcount_ == 0 && outstanding_probes_ == 0);
      VLOG(3) << "JoinBuilder (id=" << join_node_id_ << ")"
              << "closed before probe from finstance "
              << PrintId(join_node_state->fragment_instance_id())
              << "outstanding_probes_=" << outstanding_probes_ << " "
              << "probe_refcount_=" << probe_refcount_;
      DCHECK_GE(outstanding_probes_, 0);
    }
    // Only notify when there are no threads remaining
    if (no_threads_remaining) build_wakeup_cv_.NotifyAll();
  } else {
    Close(join_node_state);
  }
}

Status JoinBuilder::WaitForInitialBuild(RuntimeState* join_node_state) {
  DCHECK(is_separate_build_);
  join_node_state->AddCancellationCV(&separate_build_lock_, &probe_wakeup_cv_);
  VLOG(2) << "JoinBuilder (id=" << join_node_id_ << ")"
          << " WaitForInitialBuild() called by finstance "
          << PrintId(join_node_state->fragment_instance_id());
  unique_lock<mutex> l(separate_build_lock_);
  // Wait until either the build is ready to use or this finstance has been cancelled.
  // We can't safely pick up the build side if the build side was cancelled - instead we
  // need to wait for this finstance to be cancelled.
  while (!ready_to_probe_ && !join_node_state->is_cancelled()) {
    probe_wakeup_cv_.Wait(l);
  }
  if (join_node_state->is_cancelled()) {
    VLOG(2) << "Finstance " << PrintId(join_node_state->fragment_instance_id())
            << " cancelled while waiting for JoinBuilder (id=" << join_node_id_ << ")";
    return Status::CANCELLED;
  }
  ++probe_refcount_;
  --outstanding_probes_;
  VLOG(2) << "JoinBuilder (id=" << join_node_id_ << ")"
          << " initial build handoff to finstance "
          << PrintId(join_node_state->fragment_instance_id())
          << " probe_refcount_=" << probe_refcount_
          << " outstanding_probes_=" << outstanding_probes_;
  DCHECK_GE(outstanding_probes_, 0);
  return Status::OK();
}

void JoinBuilder::HandoffToProbesAndWait(RuntimeState* build_side_state) {
  DCHECK(is_separate_build_) << "Doesn't make sense for embedded builder.";
  VLOG(2) << "Initial build ready JoinBuilder (id=" << join_node_id_ << ")";
  build_side_state->AddCancellationCV(&separate_build_lock_, &build_wakeup_cv_);
  {
    unique_lock<mutex> l(separate_build_lock_);
    ready_to_probe_ = true;
    VLOG(3) << "JoinBuilder (id=" << join_node_id_ << ")"
            << " waiting for " << outstanding_probes_ << " probes.";
    probe_wakeup_cv_.NotifyAll();
    while (probe_refcount_ > 0
        || (outstanding_probes_ > 0 && !build_side_state->is_cancelled())) {
      SCOPED_TIMER(profile_->inactive_timer());
      VLOG(3) << "JoinBuilder (id=" << join_node_id_ << ") waiting"
              << " probe_refcount_=" << probe_refcount_
              << " outstanding_probes_=" << outstanding_probes_
              << " cancelled=" << build_side_state->is_cancelled();
      build_wakeup_cv_.Wait(l);
    }
    // Don't let probe side pick up the builder when we're going to clean it up.
    // Query cancellation will propagate to the probe finstance.
    ready_to_probe_ = !build_side_state->is_cancelled();
    VLOG(2) << "JoinBuilder (id=" << join_node_id_ << ") all probes complete. "
            << " probe_refcount_=" << probe_refcount_
            << " outstanding_probes_=" << outstanding_probes_
            << " cancelled=" << build_side_state->is_cancelled();
  }
}

void JoinBuilder::PublishRuntimeFilters(const std::vector<FilterContext>& filter_ctxs,
    RuntimeState* runtime_state, float minmax_filter_threshold, int64_t num_build_rows) {
  VLOG(3) << name() << " publishing "
          << filter_ctxs.size() << " filters.";
  int32_t num_enabled_filters = 0;
  for (const FilterContext& ctx : filter_ctxs) {
    BloomFilter* bloom_filter = nullptr;
    if (ctx.local_bloom_filter != nullptr) {
      bloom_filter = ctx.local_bloom_filter;
      ++num_enabled_filters;
    } else if (ctx.local_min_max_filter != nullptr) {
      /// Apply the column min/max stats (if applicable) to shut down the min/max
      /// filter early by setting always true flag for the filter. Do this only if
      /// the min/max filter is too close in area to the column stats of all target
      /// scan columns.
      const TRuntimeFilterDesc& filter_desc = ctx.filter->filter_desc();
      VLOG(3) << "Check out the usefulness of the local minmax filter:"
              << " id=" << ctx.filter->id()
              << ", filter details=" << ctx.local_min_max_filter->DebugString()
              << ", column stats:"
              << " low=" << PrintTColumnValue(filter_desc.targets[0].low_value)
              << ", high=" << PrintTColumnValue(filter_desc.targets[0].high_value)
              << ", threshold=" << minmax_filter_threshold
              << ", #targets=" << filter_desc.targets.size();
      bool all_overlap = true;
      for (const auto& target_desc : filter_desc.targets) {
        if (!FilterContext::ShouldRejectFilterBasedOnColumnStats(
                target_desc, ctx.local_min_max_filter, minmax_filter_threshold)) {
          all_overlap = false;
          break;
        }
      }
      if (all_overlap) {
        ctx.local_min_max_filter->SetAlwaysTrue();
        VLOG(3) << "The local minmax filter is set to always true:"
                << " id=" << ctx.filter->id();
      }

      if (!ctx.local_min_max_filter->AlwaysTrue()) {
        ++num_enabled_filters;
      }
    } else if (ctx.local_in_list_filter != nullptr) {
      if (!ctx.local_in_list_filter->AlwaysTrue()) {
        ++num_enabled_filters;
      }
    }

    runtime_state->filter_bank()->UpdateFilterFromLocal(ctx.filter->id(),
        bloom_filter, ctx.local_min_max_filter, ctx.local_in_list_filter);

    if (ctx.local_min_max_filter != nullptr) {
      VLOG(3) << name() << " published min/max filter: "
              << " id=" << ctx.filter->id()
              << ", details=" << ctx.local_min_max_filter->DebugString();
    }
  }

  if (filter_ctxs.size() > 0) {
    string info_string;
    if (num_enabled_filters == filter_ctxs.size()) {
      info_string = Substitute("$0 of $0 Runtime Filter$1 Published", filter_ctxs.size(),
          filter_ctxs.size() == 1 ? "" : "s");
    } else {
      info_string = Substitute("$0 of $1 Runtime Filter$2 Published, $3 Disabled",
          num_enabled_filters, filter_ctxs.size(), filter_ctxs.size() == 1 ? "" : "s",
          filter_ctxs.size() - num_enabled_filters);
    }
    profile()->AddInfoString("Runtime filters", info_string);
  }
}
} // namespace impala