<commit_before>// Copyright 2012 Cloudera Inc.
//
// Licensed 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/hdfs-scan-node.h"
#include "exec/base-sequence-scanner.h"
#include "exec/hdfs-text-scanner.h"
#include "exec/hdfs-lzo-text-scanner.h"
#include "exec/hdfs-sequence-scanner.h"
#include "exec/hdfs-rcfile-scanner.h"
#include "exec/hdfs-avro-scanner.h"
#include "exec/hdfs-trevni-scanner.h"

#include <sstream>
#include <boost/algorithm/string.hpp>
#include <dlfcn.h>

#include "codegen/llvm-codegen.h"
#include "common/logging.h"
#include "common/object-pool.h"
#include "exec/scan-range-context.h"
#include "exprs/expr.h"
#include "runtime/descriptors.h"
#include "runtime/hdfs-fs-cache.h"
#include "runtime/runtime-state.h"
#include "runtime/mem-pool.h"
#include "runtime/raw-value.h"
#include "runtime/row-batch.h"
#include "util/debug-util.h"
#include "util/impalad-metrics.h"
#include "util/runtime-profile.h"
#include "util/disk-info.h"

#include "gen-cpp/PlanNodes_types.h"

// TODO: temp change to validate we don't have an incast problem for joins with big tables
DEFINE_bool(randomize_scan_ranges, false, 
    "if true, randomizes the order of scan ranges");
DEFINE_int32(max_row_batches, 0, "the maximum size of materialized_row_batches_");

using namespace boost;
using namespace impala;
using namespace llvm;
using namespace std;

HdfsScanNode::HdfsScanNode(ObjectPool* pool, const TPlanNode& tnode,
                           const DescriptorTbl& descs)
    : ScanNode(pool, tnode, descs),
      thrift_plan_node_(new TPlanNode(tnode)),
      tuple_id_(tnode.hdfs_scan_node.tuple_id),
      compact_data_(tnode.compact_data),
      reader_context_(NULL),
      tuple_desc_(NULL),
      unknown_disk_id_warned_(false),
      tuple_pool_(new MemPool()),
      num_unqueued_files_(0),
      scanner_pool_(new ObjectPool()),
      num_partition_keys_(0),
      num_owned_io_buffers_(0),
      done_(false),
      partition_key_pool_(new MemPool()),
      next_range_to_issue_idx_(0),
      all_ranges_in_queue_(false),
      ranges_in_flight_(0),
      all_ranges_issued_(false),
      counters_reported_(false) {
  max_materialized_row_batches_ = FLAGS_max_row_batches;
  if (max_materialized_row_batches_ <= 0) {
    max_materialized_row_batches_ = 50 * DiskInfo::num_disks();
  }
}

HdfsScanNode::~HdfsScanNode() {
}

Status HdfsScanNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT));
  RETURN_IF_CANCELLED(state);
  SCOPED_TIMER(runtime_profile_->total_time_counter());

  {
    unique_lock<recursive_mutex> l(lock_);
    if (per_file_scan_ranges_.size() == 0 || ReachedLimit() || !status_.ok()) {
      UpdateCounters();
      *eos = true;
      return status_;
    }
  }

  RowBatch* materialized_batch = NULL;
  {
    unique_lock<mutex> l(row_batches_lock_);
    while (materialized_row_batches_.empty() && !done_) {
      row_batch_added_cv_.wait(l);
    }

    // Return any errors<commit_msg>Perf fix: revert max_row_batches to 10*num_disks<commit_after>// Copyright 2012 Cloudera Inc.
//
// Licensed 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/hdfs-scan-node.h"
#include "exec/base-sequence-scanner.h"
#include "exec/hdfs-text-scanner.h"
#include "exec/hdfs-lzo-text-scanner.h"
#include "exec/hdfs-sequence-scanner.h"
#include "exec/hdfs-rcfile-scanner.h"
#include "exec/hdfs-avro-scanner.h"
#include "exec/hdfs-trevni-scanner.h"

#include <sstream>
#include <boost/algorithm/string.hpp>
#include <dlfcn.h>

#include "codegen/llvm-codegen.h"
#include "common/logging.h"
#include "common/object-pool.h"
#include "exec/scan-range-context.h"
#include "exprs/expr.h"
#include "runtime/descriptors.h"
#include "runtime/hdfs-fs-cache.h"
#include "runtime/runtime-state.h"
#include "runtime/mem-pool.h"
#include "runtime/raw-value.h"
#include "runtime/row-batch.h"
#include "util/debug-util.h"
#include "util/impalad-metrics.h"
#include "util/runtime-profile.h"
#include "util/disk-info.h"

#include "gen-cpp/PlanNodes_types.h"

// TODO: temp change to validate we don't have an incast problem for joins with big tables
DEFINE_bool(randomize_scan_ranges, false, 
    "if true, randomizes the order of scan ranges");
DEFINE_int32(max_row_batches, 0, "the maximum size of materialized_row_batches_");

using namespace boost;
using namespace impala;
using namespace llvm;
using namespace std;

HdfsScanNode::HdfsScanNode(ObjectPool* pool, const TPlanNode& tnode,
                           const DescriptorTbl& descs)
    : ScanNode(pool, tnode, descs),
      thrift_plan_node_(new TPlanNode(tnode)),
      tuple_id_(tnode.hdfs_scan_node.tuple_id),
      compact_data_(tnode.compact_data),
      reader_context_(NULL),
      tuple_desc_(NULL),
      unknown_disk_id_warned_(false),
      tuple_pool_(new MemPool()),
      num_unqueued_files_(0),
      scanner_pool_(new ObjectPool()),
      num_partition_keys_(0),
      num_owned_io_buffers_(0),
      done_(false),
      partition_key_pool_(new MemPool()),
      next_range_to_issue_idx_(0),
      all_ranges_in_queue_(false),
      ranges_in_flight_(0),
      all_ranges_issued_(false),
      counters_reported_(false) {
  max_materialized_row_batches_ = FLAGS_max_row_batches;
  if (max_materialized_row_batches_ <= 0) {
    // TODO: This parameter has an U-shaped effect on performance: increasing the value
    // would first improves performance, but further increasing would degrade performance.
    // Investigate and tune this.
    max_materialized_row_batches_ = 10 * DiskInfo::num_disks();
  }
}

HdfsScanNode::~HdfsScanNode() {
}

Status HdfsScanNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT));
  RETURN_IF_CANCELLED(state);
  SCOPED_TIMER(runtime_profile_->total_time_counter());

  {
    unique_lock<recursive_mutex> l(lock_);
    if (per_file_scan_ranges_.size() == 0 || ReachedLimit() || !status_.ok()) {
      UpdateCounters();
      *eos = true;
      return status_;
    }
  }

  RowBatch* materialized_batch = NULL;
  {
    unique_lock<mutex> l(row_batches_lock_);
    while (materialized_row_batches_.empty() && !done_) {
      row_batch_added_cv_.wait(l);
    }

    // Return any errors
{'fraction_non_alphanumeric': 0.12048524951750758, 'fraction_numerical': 0.005789909015715467, 'mean_word_length': 4.249638205499276, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1182318', 'n_tokens_mistral': 2506, 'n_tokens_neox': 2452, 'n_words': 546}