<commit_before>#include "dfadb_table.hpp"
#include "dfadb_segment.hpp"
#include <terark/fsa/create_regex_dfa.hpp>
#include <terark/fsa/dense_dfa.hpp>
#include <terark/io/FileStream.hpp>
#include <terark/io/DataIO.hpp>
#include <terark/num_to_str.hpp>
#include <terark/util/mmap.hpp>
#include <terark/db/appendonly.hpp>
#include <terark/db/mock_db_engine.hpp>
#include <terark/db/wiredtiger/wt_db_segment.hpp>
#include <terark/db/dfadb/nlt_index.hpp>
#include <boost/filesystem.hpp>

namespace terark { namespace db { namespace dfadb {


DfaDbContext::DfaDbContext(const CompositeTable* tab) : DbContext(tab) {
}
DfaDbContext::~DfaDbContext() {
}

DbContext* DfaDbTable::createDbContextNoLock() const {
	return new DfaDbContext(this);
}

ReadonlySegment*
DfaDbTable::createReadonlySegment(PathRef dir) const {
	std::unique_ptr<DfaDbReadonlySegment> seg(new DfaDbReadonlySegment());
	return seg.release();
}

WritableSegment*
DfaDbTable::createWritableSegment(PathRef dir) const {
	const char* dfaWritableSeg = getenv("TerarkDB_DfaWritableSegment");
	if (dfaWritableSeg && strcasecmp(dfaWritableSeg, "mock") == 0) {
		std::unique_ptr<WritableSegment> seg(new MockWritableSegment(dir));
		seg->m_schema = this->m_schema;
		return seg.release();
	}
	else {
		using terark::db::wt::WtWritableSegment;
		std::unique_ptr<WtWritableSegment> seg(new WtWritableSegment());
		seg->m_schema = this->m_schema;
		seg->load(dir);
		return seg.release();
	}
}

WritableSegment*
DfaDbTable::openWritableSegment(PathRef dir) const {
	auto isDelPath = dir / "IsDel";
	if (boost::filesystem::exists(isDelPath)) {
		const char* dfaWritableSeg = getenv("TerarkDB_DfaWritableSegment");
		if (dfaWritableSeg && strcasecmp(dfaWritableSeg, "mock") == 0) {
			std::unique_ptr<WritableSegment> seg(new MockWritableSegment(dir));
			seg->m_schema = this->m_schema;
			seg->load(dir);
			return seg.release();
		}
		else {
			using terark::db::wt::WtWritableSegment;
			std::unique_ptr<WtWritableSegment> seg(new WtWritableSegment());
			seg->m_schema = this->m_schema;
			seg->load(dir);
			return seg.release();
		}
	}
	else {
		return myCreateWritableSegment(dir);
	}
}

class AdapterRegexDFA : public DenseDFA_uint32_320 {
public:
	typedef AdapterRegexDFA MyType;
#include <terark/fsa/ppi/match_path.hpp>
};

bool
DfaDbTable::indexMatchRegex(size_t indexId, BaseDFA* regexDFA,
							valvec<llong>* recIdvec, DbContext* ctx)
const {
	if (indexId >= m_schema->getIndexNum()) {
		THROW_STD(invalid_argument
			, "invalid indexId=%zd is not less than indexNum=%zd"
			, indexId, m_schema->getIndexNum());
	}
	const Schema& schema = m_schema->getIndexSchema(indexId);
	if (schema.columnNum() > 1) {
		THROW_STD(invalid_argument
			, "can not MatchRegex on composite indexId=%zd indexName=%s"
			, indexId, schema.m_name.c_str());
	}
	auto& colmeta = schema.getColumnMeta(0);
	if (!colmeta.isString()) {
		THROW_STD(invalid_argument
			, "can not MatchRegex on non-string indexId=%zd indexName=%s"
			, indexId, schema.m_name.c_str());
	}
	recIdvec->erase_all();
	MyRwLock lock(this->m_rwMutex, false);
	for (size_t i = 0; i < m_segments.size(); ++i) {
		auto seg = m_segments[i].get();
		if (seg->getWritableStore()) {
			if (seg->m_isDel.size() > 0) {
			  fprintf(stderr
				, "WARN: segment: %s is a writable segment, can not MatchRegex\n"
				, getSegPath("wr", i).string().c_str());
			}
			continue;
		}
		auto index = dynamic_cast<const NestLoudsTrieIndex*>(&*seg->m_indices[indexId]);
		if (!index) {
			THROW_STD(logic_error, "MatchRegex must be run on NestLoudsTrieIndex\n");
		}
		size_t oldsize = recIdvec->size();
		if (index->matchRegexAppend(regexDFA, recIdvec, ctx)) {
			llong baseId = m_rowNumVec[i];
			size_t i = oldsize;
			for(size_t j = oldsize; j < recIdvec->size(); ++j) {
				size_t subPhysicId = (*recIdvec)[j];
				size_t subLogicId = seg->getLogicId(subPhysicId);
				if (!seg->m_isDel[subLogicId])
					(*recIdvec)[i++] = baseId + subLogicId;
			}
			recIdvec->risk_set_size(i);
		}
		else if (schema.m_enableLinearScan) {
			fprintf(stderr
				, "WARN: RegexMatch exceeded memory limit(%zd bytes) on index '%s' of segment: '%s', try linear scan...\n"
				, ctx->regexMatchMemLimit
				, schema.m_name.c_str(), seg->m_segDir.string().c_str());
			auto matchDFA = static_cast<const AdapterRegexDFA*>(
					dynamic_cast<const DenseDFA_uint32_320*>(regexDFA)
				);
			assert(NULL != matchDFA);
			valvec<byte> key;
			size_t subPhysicId = 0;
			size_t subLogicId = 0;
			size_t subRowsNum = seg->m_isDel.size();
			boost::intrusive_ptr<SeqReadAppendonlyStore>
				seqStore(new SeqReadAppendonlyStore(seg->m_segDir, schema));
			StoreIteratorPtr iter = seqStore->createStoreIterForward(ctx);
			const bm_uint_t* isDel = seg->m_isDel.bldata();
			const bm_uint_t* isPurged = seg->m_isPurged.bldata();
			const llong baseId = m_rowNumVec[i];
			for (; subLogicId < subRowsNum; subLogicId++) {
				if (!isPurged || !terark_bit_test(isPurged, subLogicId)) {
					llong subCheckPhysicId = INT_MAX; // for fail fast
					bool hasData = iter->increment(&subCheckPhysicId, &key);
					TERARK_RT_assert(hasData, std::logic_error);
					TERARK_RT_assert(size_t(subCheckPhysicId) == subPhysicId, std::logic_error);
					if (!terark_bit_test(isDel, subLogicId)) {
						if (matchDFA->first_mismatch_pos(key) == key.size()) {
							recIdvec->push_back(baseId + subLogicId);
						}
					}<commit_msg>dfadb_table.cpp: Fix for inprogress-snapshot<commit_after>#include "dfadb_table.hpp"
#include "dfadb_segment.hpp"
#include <terark/fsa/create_regex_dfa.hpp>
#include <terark/fsa/dense_dfa.hpp>
#include <terark/io/FileStream.hpp>
#include <terark/io/DataIO.hpp>
#include <terark/num_to_str.hpp>
#include <terark/util/mmap.hpp>
#include <terark/db/appendonly.hpp>
#include <terark/db/mock_db_engine.hpp>
#include <terark/db/wiredtiger/wt_db_segment.hpp>
#include <terark/db/dfadb/nlt_index.hpp>
#include <boost/filesystem.hpp>

namespace terark { namespace db { namespace dfadb {


DfaDbContext::DfaDbContext(const CompositeTable* tab) : DbContext(tab) {
}
DfaDbContext::~DfaDbContext() {
}

DbContext* DfaDbTable::createDbContextNoLock() const {
	return new DfaDbContext(this);
}

ReadonlySegment*
DfaDbTable::createReadonlySegment(PathRef dir) const {
	std::unique_ptr<DfaDbReadonlySegment> seg(new DfaDbReadonlySegment());
	return seg.release();
}

WritableSegment*
DfaDbTable::createWritableSegment(PathRef dir) const {
	const char* dfaWritableSeg = getenv("TerarkDB_DfaWritableSegment");
	if (dfaWritableSeg && strcasecmp(dfaWritableSeg, "mock") == 0) {
		std::unique_ptr<WritableSegment> seg(new MockWritableSegment(dir));
		seg->m_schema = this->m_schema;
		return seg.release();
	}
	else {
		using terark::db::wt::WtWritableSegment;
		std::unique_ptr<WtWritableSegment> seg(new WtWritableSegment());
		seg->m_schema = this->m_schema;
		seg->load(dir);
		return seg.release();
	}
}

WritableSegment*
DfaDbTable::openWritableSegment(PathRef dir) const {
	auto isDelPath = dir / "IsDel";
	if (boost::filesystem::exists(isDelPath)) {
		const char* dfaWritableSeg = getenv("TerarkDB_DfaWritableSegment");
		if (dfaWritableSeg && strcasecmp(dfaWritableSeg, "mock") == 0) {
			std::unique_ptr<WritableSegment> seg(new MockWritableSegment(dir));
			seg->m_schema = this->m_schema;
			seg->load(dir);
			return seg.release();
		}
		else {
			using terark::db::wt::WtWritableSegment;
			std::unique_ptr<WtWritableSegment> seg(new WtWritableSegment());
			seg->m_schema = this->m_schema;
			seg->load(dir);
			return seg.release();
		}
	}
	else {
		return myCreateWritableSegment(dir);
	}
}

class AdapterRegexDFA : public DenseDFA_uint32_320 {
public:
	typedef AdapterRegexDFA MyType;
#include <terark/fsa/ppi/match_path.hpp>
};

bool
DfaDbTable::indexMatchRegex(size_t indexId, BaseDFA* regexDFA,
							valvec<llong>* recIdvec, DbContext* ctx)
const {
	if (indexId >= m_schema->getIndexNum()) {
		THROW_STD(invalid_argument
			, "invalid indexId=%zd is not less than indexNum=%zd"
			, indexId, m_schema->getIndexNum());
	}
	const Schema& schema = m_schema->getIndexSchema(indexId);
	if (schema.columnNum() > 1) {
		THROW_STD(invalid_argument
			, "can not MatchRegex on composite indexId=%zd indexName=%s"
			, indexId, schema.m_name.c_str());
	}
	auto& colmeta = schema.getColumnMeta(0);
	if (!colmeta.isString()) {
		THROW_STD(invalid_argument
			, "can not MatchRegex on non-string indexId=%zd indexName=%s"
			, indexId, schema.m_name.c_str());
	}
	ctx->trySyncSegCtxSpeculativeLock(this);
	recIdvec->erase_all();
	for (size_t i = 0; i < ctx->m_segCtx.size(); ++i) {
		auto seg = ctx->m_segCtx[i]->seg;
		if (seg->getWritableStore()) {
			if (seg->m_isDel.size() > 0) {
			  fprintf(stderr
				, "WARN: segment: %s is a writable segment, can not MatchRegex\n"
				, getSegPath("wr", i).string().c_str());
			}
			continue;
		}
		auto index = dynamic_cast<const NestLoudsTrieIndex*>(&*seg->m_indices[indexId]);
		if (!index) {
			THROW_STD(logic_error, "MatchRegex must be run on NestLoudsTrieIndex\n");
		}
		size_t oldsize = recIdvec->size();
		const llong* deltime = nullptr;
		const llong  baseId = ctx->m_rowNumVec[i];
		llong snapshotVersion = ctx->m_mySnapshotVersion;
		if (seg->m_deletionTime) {
			assert(nullptr != m_schema->m_snapshotSchema);
			deltime = (const llong*)(seg->m_deletionTime->getRecordsBasePtr());
		}
		if (index->matchRegexAppend(regexDFA, recIdvec, ctx)) {
			size_t i = oldsize;
			for(size_t j = oldsize; j < recIdvec->size(); ++j) {
				size_t subPhysicId = (*recIdvec)[j];
				size_t subLogicId = seg->getLogicId(subPhysicId);
				if (deltime) {
					if (deltime[subPhysicId] > snapshotVersion)
						(*recIdvec)[i++] = baseId + subLogicId;
				}
				else {
					if (!seg->m_isDel[subLogicId])
						(*recIdvec)[i++] = baseId + subLogicId;
				}
			}
			recIdvec->risk_set_size(i);
		}
		else if (schema.m_enableLinearScan) {
			fprintf(stderr
				, "WARN: RegexMatch exceeded memory limit(%zd bytes) on index '%s' of segment: '%s', try linear scan...\n"
				, ctx->regexMatchMemLimit
				, schema.m_name.c_str(), seg->m_segDir.string().c_str());
			auto matchDFA = static_cast<const AdapterRegexDFA*>(
					dynamic_cast<const DenseDFA_uint32_320*>(regexDFA)
				);
			assert(NULL != matchDFA);
			valvec<byte> key;
			size_t subPhysicId = 0;
			size_t subLogicId = 0;
			size_t subRowsNum = seg->m_isDel.size();
			boost::intrusive_ptr<SeqReadAppendonlyStore>
				seqStore(new SeqReadAppendonlyStore(seg->m_segDir, schema));
			StoreIteratorPtr iter = seqStore->createStoreIterForward(ctx);
			const bm_uint_t* isDel = seg->m_isDel.bldata();
			const bm_uint_t* isPurged = seg->m_isPurged.bldata();
			for (; subLogicId < subRowsNum; subLogicId++) {
				if (!isPurged || !terark_bit_test(isPurged, subLogicId)) {
					llong subCheckPhysicId = INT_MAX; // for fail fast
					bool hasData = iter->increment(&subCheckPhysicId, &key);
					TERARK_RT_assert(hasData, std::logic_error);
					TERARK_RT_assert(size_t(subCheckPhysicId) == subPhysicId, std::logic_error);
					if (deltime) {
						if (deltime[subPhysicId] > snapshotVersion) {
							if (matchDFA->first_mismatch_pos(key) == key.size()) {
								recIdvec->push_back(baseId + subLogicId);
							}
						}
					}
					else {
						if (!terark_bit_test(isDel, subLogicId)) {
							if (matchDFA->first_mismatch_pos(key) == key.size()) {
								recIdvec->push_back(baseId + subLogicId);
							}
						}
					}
{'fraction_non_alphanumeric': 0.14442493415276558, 'fraction_numerical': 0.003160667251975417, 'mean_word_length': 4.811734693877551, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 138, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6701572', 'n_tokens_mistral': 4809, 'n_tokens_neox': 4288, 'n_words': 746}