<commit_before>#include "sort_file.h"
#include <logging.h>

using baidu::common::Log;
using baidu::common::FATAL;
using baidu::common::INFO;
using baidu::common::WARNING;

namespace baidu {
namespace shuttle {

MergeFileReader::~MergeFileReader() {
    std::vector<SortFileReader*>::iterator it;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        delete (*it);
    }
}

Status MergeFileReader::Open(const std::vector<std::string>& files, 
                             FileSystem::Param param,
                             FileType file_type) {
    if (files.size() == 0) {
        return kInvalidArg;
    }
    Status status = kOk;

    std::vector<std::string>::const_iterator it;
    for (it = files.begin(); it != files.end(); it++) {
        const std::string& file_name = *it;
        Status st;
        SortFileReader* reader = SortFileReader::Create(file_type, &st);
        if (st == kOk) {
            st = reader->Open(file_name, param);
        }
        if (st != kOk) {
            status = st;
            LOG(WARNING, "failed to open %s, status: %s", 
                file_name.c_str(), Status_Name(st).c_str());
            err_file_ = file_name;
            break;
        }
        readers_.push_back(reader);
    }
    
    return status;
}

Status MergeFileReader::Close() {
    std::vector<SortFileReader*>::iterator it;
    Status status = kOk;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        Status st = (*it)->Close();
        if (st != kOk) {
            status = st;
            err_file_ = (*it)->GetFileName();
            break;
        }
    }
    return status;
}

SortFileReader::Iterator* MergeFileReader::Scan(const std::string& start_key, const std::string& end_key) {
    std::vector<SortFileReader::Iterator*> iters;
    std::vector<SortFileReader*>::iterator it;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        SortFileReader * const& reader = *it;
        iters.push_back(reader->Scan(start_key, end_key));        
    }
    return new MergeIterator(iters, this);
}

MergeFileReader::MergeIterator::MergeIterator(const std::vector<SortFileReader::Iterator*>& iters,
                                              MergeFileReader* reader) {
    merge_reader_ = reader;
    std::vector<SortFileReader::Iterator*>::const_iterator it;
    status_ = kOk;
    int offset = 0;
    for (it = iters.begin(); it != iters.end(); it++) {
        SortFileReader::Iterator * const& reader_it = *it;
        if (!reader_it->Done()) {
            queue_.push(MergeItem(reader_it->Key(), reader_it->Value(), offset));
            iters_.push_back(reader_it);
            offset++;
        }
        if (reader_it->Error() != kOk) {
            status_ = reader_it->Error();
            merge_reader_->err_file_ = reader_it->GetFileName();
        }
    }
    if (status_ == kOk && !queue_.empty()) {
        key_ = queue_.top().key_;
        value_ = queue_.top().value_;
    }
}

MergeFileReader::MergeIterator::~MergeIterator() {
    std::vector<SortFileReader::Iterator*>::iterator it;
    for (it = iters_.begin(); it != iters_.end(); it++) {<commit_msg>bug fix: fix memory leak of shuffle tool<commit_after>#include "sort_file.h"
#include <logging.h>

using baidu::common::Log;
using baidu::common::FATAL;
using baidu::common::INFO;
using baidu::common::WARNING;

namespace baidu {
namespace shuttle {

MergeFileReader::~MergeFileReader() {
    std::vector<SortFileReader*>::iterator it;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        delete (*it);
    }
}

Status MergeFileReader::Open(const std::vector<std::string>& files, 
                             FileSystem::Param param,
                             FileType file_type) {
    if (files.size() == 0) {
        return kInvalidArg;
    }
    Status status = kOk;

    std::vector<std::string>::const_iterator it;
    for (it = files.begin(); it != files.end(); it++) {
        const std::string& file_name = *it;
        Status st;
        SortFileReader* reader = SortFileReader::Create(file_type, &st);
        if (st == kOk) {
            st = reader->Open(file_name, param);
        }
        if (st != kOk) {
            status = st;
            LOG(WARNING, "failed to open %s, status: %s", 
                file_name.c_str(), Status_Name(st).c_str());
            err_file_ = file_name;
            break;
        }
        readers_.push_back(reader);
    }
    
    return status;
}

Status MergeFileReader::Close() {
    std::vector<SortFileReader*>::iterator it;
    Status status = kOk;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        Status st = (*it)->Close();
        if (st != kOk) {
            status = st;
            err_file_ = (*it)->GetFileName();
            break;
        }
    }
    return status;
}

SortFileReader::Iterator* MergeFileReader::Scan(const std::string& start_key, const std::string& end_key) {
    std::vector<SortFileReader::Iterator*> iters;
    std::vector<SortFileReader*>::iterator it;
    for (it = readers_.begin(); it != readers_.end(); it++) {
        SortFileReader * const& reader = *it;
        iters.push_back(reader->Scan(start_key, end_key));        
    }
    return new MergeIterator(iters, this);
}

MergeFileReader::MergeIterator::MergeIterator(const std::vector<SortFileReader::Iterator*>& iters,
                                              MergeFileReader* reader) {
    merge_reader_ = reader;
    std::vector<SortFileReader::Iterator*>::const_iterator it;
    status_ = kOk;
    int offset = 0;
    for (it = iters.begin(); it != iters.end(); it++) {
        SortFileReader::Iterator * const& reader_it = *it;
        bool drained = false;
        if (!reader_it->Done()) {
            queue_.push(MergeItem(reader_it->Key(), reader_it->Value(), offset));
            iters_.push_back(reader_it);
            offset++;
        } else {
            drained = true;
        }
        if (reader_it->Error() != kOk) {
            status_ = reader_it->Error();
            merge_reader_->err_file_ = reader_it->GetFileName();
        }
        if (drained) {
            delete reader_it;
        }
    }
    if (status_ == kOk && !queue_.empty()) {
        key_ = queue_.top().key_;
        value_ = queue_.top().value_;
    }
}

MergeFileReader::MergeIterator::~MergeIterator() {
    std::vector<SortFileReader::Iterator*>::iterator it;
    for (it = iters_.begin(); it != iters_.end(); it++) {
{'fraction_non_alphanumeric': 0.1535121724298341, 'fraction_numerical': 0.0006202512017367033, 'mean_word_length': 2.4126984126984126, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3487931', 'n_tokens_mistral': 1972, 'n_tokens_neox': 1959, 'n_words': 407}