lszskye commented on code in PR #490:
URL: https://github.com/apache/paimon-cpp/pull/490#discussion_r4476460546
##########
src/paimon/core/io/file_index_evaluator.cpp:
##########
@@ -52,64 +54,165 @@ namespace paimon {
class MemoryPool;
enum class FieldType;
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
+FileIndexEvaluator::FileIndexEvaluator(std::shared_ptr<arrow::Schema>
data_schema,
+ std::shared_ptr<DataFileMeta> file_meta,
+
std::unique_ptr<FileIndexFormat::Reader> format_reader)
+ : data_schema_(std::move(data_schema)),
+ file_meta_(std::move(file_meta)),
+ format_reader_(std::move(format_reader)) {}
+
+Result<std::unique_ptr<FileIndexEvaluator>> FileIndexEvaluator::Create(
const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& options,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
const std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/false, data_schema, options,
predicate,
- data_file_path_factory, file_meta, file_system, pool);
+ std::unique_ptr<FileIndexFormat::Reader> format_reader;
+ if (file_meta->row_count >= 0 && file_meta->row_count <=
RoaringBitmap32::MAX_VALUE) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+ ExtractIndexInputStream(path_factory,
file_meta, file_system));
+ if (input_stream) {
+ PAIMON_ASSIGN_OR_RAISE(format_reader,
+ FileIndexFormat::CreateReader(input_stream,
pool, options));
+ }
+ }
+ return std::unique_ptr<FileIndexEvaluator>(
+ new FileIndexEvaluator(data_schema, file_meta,
std::move(format_reader)));
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/true, data_schema, options,
predicate,
- /*data_file_path_factory=*/nullptr, file_meta,
/*file_system=*/nullptr, pool);
+Result<std::vector<std::shared_ptr<FileIndexReader>>>
FileIndexEvaluator::GetReaders(
+ const std::string& field_name) {
+ auto iter = field_readers_.find(field_name);
+ if (iter != field_readers_.end()) {
+ return iter->second;
+ }
+ std::shared_ptr<arrow::Field> field =
data_schema_->GetFieldByName(field_name);
+ if (!field) {
+ return Status::Invalid(fmt::format("field '{}' does not exist in data
schema", field_name));
+ }
+ std::vector<std::shared_ptr<FileIndexReader>> readers;
+ if (format_reader_) {
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportSchema(*arrow::schema({std::move(field)}),
&c_schema));
+ PAIMON_ASSIGN_OR_RAISE(readers,
format_reader_->ReadColumnIndex(field_name, &c_schema));
+ }
+ field_readers_.emplace(field_name, readers);
+ return readers;
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- bool only_use_embedding_index, const std::shared_ptr<arrow::Schema>&
data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
- const std::shared_ptr<MemoryPool>& pool) {
- if (predicate == nullptr || file_meta->row_count >
RoaringBitmap32::MAX_VALUE) {
- return FileIndexResult::Remain();
+Result<std::shared_ptr<FileIndexReader>> FileIndexEvaluator::GetSearchReader(
+ const std::string& field_name, FileIndexReader::SearchType search_type) {
+ if (!data_schema_->GetFieldByName(field_name)) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' does not exist in data schema",
field_name));
+ }
+ if (!format_reader_) {
+ return Status::Invalid(fmt::format("Data file '{}' has no File Index
for search field '{}'",
+ file_meta_->file_name, field_name));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(field_name));
+ std::shared_ptr<FileIndexReader> selected;
+ for (const auto& reader : readers) {
+ if (!reader->SupportsSearch(search_type)) {
+ continue;
+ }
+ if (selected) {
+ return Status::Invalid(fmt::format(
+ "Multiple File Index readers support requested search for
field '{}'", field_name));
+ }
+ selected = reader;
}
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
- ExtractIndexInputStream(only_use_embedding_index,
data_file_path_factory,
- file_meta, file_system));
- if (!input_stream) {
- // no index files, just return REMAIN
+ if (!selected) {
+ return Status::NotImplemented(fmt::format(
+ "No File Index reader supports requested search for field '{}'",
field_name));
+ }
+ return selected;
+}
+
+Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::EvaluateLeaf(
+ const std::shared_ptr<LeafPredicate>& predicate, bool needs_scores) {
+ if (predicate->GetFunction().GetType() == Function::Type::FULL_TEXT_MATCH)
{
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexReader> reader,
+ GetSearchReader(predicate->FieldName(),
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ auto full_text =
std::dynamic_pointer_cast<FullTextMatchPredicate>(predicate);
+ if (!full_text) {
+ return Status::Invalid("FullTextMatch requires a full-text match
predicate");
+ }
+ return reader->VisitFullTextMatch(full_text->GetFullTextMatch(),
needs_scores);
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(predicate->FieldName()));
+ std::shared_ptr<FileIndexResult> result = FileIndexResult::Remain();
+ for (const auto& reader : readers) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexResult> child,
+
PredicateUtils::VisitPredicate<std::shared_ptr<FileIndexResult>>(predicate,
reader));
+ PAIMON_ASSIGN_OR_RAISE(result, result->And(child));
+ PAIMON_ASSIGN_OR_RAISE(bool remain, result->IsRemain());
Review Comment:
Thanks for catching this. This is a pre-existing issue in Bloom index, this
PR does not change that logic. I’ll address this in a separate bug-fix PR with
a regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
