This is an automated email from the ASF dual-hosted git repository.
airborne12 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cab48ddc68a [refactor](be) Add the CLucene-free text analysis
interface and its two CLucene adapters (#70261)
cab48ddc68a is described below
commit cab48ddc68abd4d75d3d8729ea9b6e9c66699c9a
Author: Jack <[email protected]>
AuthorDate: Fri Nov 6 22:55:29 2026 +0800
[refactor](be) Add the CLucene-free text analysis interface and its two
CLucene adapters (#70261)
### What problem does this PR solve?
Issue Number: None
Related PR: #70260 (depends on)
Problem Summary: Add a CLucene-free text analysis interface
(`TextAnalyzer`, `TextTokenStream`, `TokenBatch`) that hands out tokens
in batches, and adapters in both directions so that consumers and
analyzers can move to it separately. Nothing in production uses it yet.
---
.../index/inverted/analysis/per_token_stream.cpp | 84 ++++
.../index/inverted/analysis/per_token_stream.h | 77 ++++
be/src/storage/index/inverted/analysis/scratch.h | 39 ++
.../index/inverted/analysis/text_analyzer.h | 39 ++
.../index/inverted/analysis/text_token_stream.cpp | 42 ++
.../index/inverted/analysis/text_token_stream.h | 89 ++++
.../storage/index/inverted/analysis/token_batch.h | 102 +++++
.../storage/index/inverted/analysis/token_stream.h | 13 +-
.../inverted/clucene/lucene_analyzer_bridge.cpp | 111 +++++
.../inverted/clucene/lucene_analyzer_bridge.h | 88 ++++
.../clucene/lucene_backed_text_analyzer.cpp | 117 +++++
.../inverted/clucene/lucene_backed_text_analyzer.h | 53 +++
.../analysis/golden/analysis_golden_test.cpp | 125 +++++-
.../analysis/golden/analysis_write_golden_test.cpp | 150 ++++++-
.../inverted/analysis/golden/golden_driver.cpp | 104 ++++-
.../index/inverted/analysis/golden/golden_driver.h | 12 +
.../inverted/analysis/text_stream_contract.cpp | 80 ++++
.../index/inverted/analysis/text_stream_contract.h | 56 +++
.../inverted/analysis/text_token_stream_test.cpp | 471 +++++++++++++++++++++
.../index/inverted/clucene/lucene_adapter_test.cpp | 455 ++++++++++++++++++++
20 files changed, 2262 insertions(+), 45 deletions(-)
diff --git a/be/src/storage/index/inverted/analysis/per_token_stream.cpp
b/be/src/storage/index/inverted/analysis/per_token_stream.cpp
new file mode 100644
index 00000000000..8ce5fe17a98
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/per_token_stream.cpp
@@ -0,0 +1,84 @@
+// 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 "storage/index/inverted/analysis/per_token_stream.h"
+
+namespace doris::segment_v2::inverted_index {
+
+namespace {
+
+// Copies tokens into `batch` until it is full or `next_token` returns false,
which latches
+// `exhausted`.
+template <typename NextToken>
+bool fill_batch(TokenBatch& batch, bool& exhausted, NextToken&& next_token) {
+ batch.clear();
+ TokenRef token;
+ while (!exhausted && batch.size() < TextTokenStream::kBatchTokens) {
+ if (next_token(token)) {
+ batch.push_copy(token.term(), token.position_increment);
+ } else {
+ exhausted = true;
+ }
+ }
+ return batch.size() != 0;
+}
+
+} // namespace
+
+void PerTokenTokenizer::reset(std::string_view text) {
+ // Exhausted until on_reset() returns, so a stream whose on_reset() threw
makes no token.
+ _exhausted = true;
+ on_reset(text);
+ _exhausted = false;
+}
+
+bool PerTokenTokenizer::next_batch(TokenBatch& batch) {
+ return fill_batch(batch, _exhausted, [this](TokenRef& token) { return
next_token(token); });
+}
+
+void PerTokenFilter::reset(std::string_view text) {
+ _exhausted = true;
+ TextTokenFilter::reset(text);
+ // A half-read value leaves tokens in the input batch; none of them may
come out again.
+ _input.clear();
+ _input.release_oversized();
+ _cursor = 0;
+ on_reset();
+ _exhausted = false;
+}
+
+bool PerTokenFilter::next_batch(TokenBatch& batch) {
+ return fill_batch(batch, _exhausted, [this](TokenRef& token) { return
next_token(token); });
+}
+
+size_t PerTokenFilter::reserved_bytes() const {
+ return _input.capacity_bytes() + TextTokenFilter::reserved_bytes();
+}
+
+bool PerTokenFilter::input_token(TokenRef& token) {
+ if (_cursor == _input.size()) {
+ _cursor = 0;
+ // An exhausted input keeps returning false with an empty batch.
+ if (!_in->next_batch(_input)) {
+ return false;
+ }
+ }
+ token = _input.tokens()[_cursor++];
+ return true;
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/per_token_stream.h
b/be/src/storage/index/inverted/analysis/per_token_stream.h
new file mode 100644
index 00000000000..e46357b082b
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/per_token_stream.h
@@ -0,0 +1,77 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <string_view>
+
+#include "storage/index/inverted/analysis/text_token_stream.h"
+
+namespace doris::segment_v2::inverted_index {
+
+// The base of a tokenizer that makes one token at a time. It keeps its state
machine and may build
+// every term in one reused scratch buffer: the base copies each term into the
batch, because views
+// of a reused scratch would overwrite each other within one batch.
+class PerTokenTokenizer : public TextTokenizer {
+public:
+ void reset(std::string_view text) final;
+ bool next_batch(TokenBatch& batch) final;
+
+protected:
+ // Binds `text`, dropping the state of the previous value.
+ virtual void on_reset(std::string_view text) = 0;
+ // The next token; its term stays valid until the next call. Once it
returned false it is not
+ // called again before the next reset().
+ virtual bool next_token(TokenRef& token) = 0;
+
+private:
+ // No value is bound, or next_token() returned false.
+ bool _exhausted = true;
+};
+
+// The base of a filter that makes one token at a time out of the tokens of
its input, which it
+// reads with input_token(). The input batch is private and only input_token()
refills it, so the
+// filter may hold an input token across next_batch() calls, as word_delimiter
does while it hands
+// out the parts of one token.
+class PerTokenFilter : public TextTokenFilter {
+public:
+ using TextTokenFilter::TextTokenFilter;
+
+ void reset(std::string_view text) final;
+ bool next_batch(TokenBatch& batch) final;
+ // Adds the input batch. A filter with buffers of its own overrides it and
adds them.
+ size_t reserved_bytes() const override;
+
+protected:
+ // Drops the filter's state of the previous value; the input is reset
already.
+ virtual void on_reset() {}
+ // The next token; its term stays valid until the next call. Once it
returned false it is not
+ // called again before the next reset().
+ virtual bool next_token(TokenRef& token) = 0;
+ // The next input token, or false once the input is exhausted. Its term
stays valid until the
+ // next call.
+ bool input_token(TokenRef& token);
+
+private:
+ TokenBatch _input;
+ size_t _cursor = 0; // the next token of _input
+ // No value is bound, or next_token() returned false.
+ bool _exhausted = true;
+};
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/scratch.h
b/be/src/storage/index/inverted/analysis/scratch.h
new file mode 100644
index 00000000000..68fffdfdbd8
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/scratch.h
@@ -0,0 +1,39 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+
+namespace doris::segment_v2::inverted_index {
+
+// The capacity an analysis buffer may keep from one value to the next. Values
have no length
+// limit, so a buffer that grew for a long value is released once the value is
done; otherwise a
+// stream reused across values would hold, say, 800 MB after one 50 MB value
until the segment
+// ends. 64 KiB is far above what ordinary values need, so steady values
allocate nothing.
+constexpr size_t ANALYZER_SCRATCH_HIGH_WATER_BYTES = 64 * 1024;
+
+// Releases `container` when its capacity is above the high water mark.
+template <typename Container>
+void release_oversized_scratch(Container& container) {
+ if (container.capacity() * sizeof(typename Container::value_type) >
+ ANALYZER_SCRATCH_HIGH_WATER_BYTES) {
+ Container().swap(container);
+ }
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/text_analyzer.h
b/be/src/storage/index/inverted/analysis/text_analyzer.h
new file mode 100644
index 00000000000..08d20061844
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/text_analyzer.h
@@ -0,0 +1,39 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+
+#include "storage/index/inverted/analysis/text_token_stream.h"
+
+namespace doris::segment_v2::inverted_index {
+
+// Makes the token streams of one analysis configuration. An analyzer does not
change once it is
+// built, so any number of threads may share one, each with streams of its own.
+class TextAnalyzer {
+public:
+ virtual ~TextAnalyzer() = default;
+
+ // A new stream. It must not outlive the analyzer: a stream may refer to
what the analyzer
+ // holds, as an ICU normalizer refers to the character set its factory
keeps.
+ virtual std::unique_ptr<TextTokenStream> create_stream() const = 0;
+};
+
+using TextAnalyzerPtr = std::shared_ptr<const TextAnalyzer>;
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/text_token_stream.cpp
b/be/src/storage/index/inverted/analysis/text_token_stream.cpp
new file mode 100644
index 00000000000..154ee63cd10
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/text_token_stream.cpp
@@ -0,0 +1,42 @@
+// 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 "storage/index/inverted/analysis/text_token_stream.h"
+
+#include <utility>
+
+#include "common/check.h"
+
+namespace doris::segment_v2::inverted_index {
+
+TextTokenFilter::TextTokenFilter(std::unique_ptr<TextTokenStream> in) :
_in(std::move(in)) {
+ DORIS_CHECK(_in != nullptr);
+}
+
+void TextTokenFilter::reset(std::string_view text) {
+ _in->reset(text);
+}
+
+size_t TextTokenFilter::reserved_bytes() const {
+ return _in->reserved_bytes();
+}
+
+void TextTokenFilter::set_strict_utf8(bool strict) {
+ _in->set_strict_utf8(strict);
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/text_token_stream.h
b/be/src/storage/index/inverted/analysis/text_token_stream.h
new file mode 100644
index 00000000000..1198209cdca
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/text_token_stream.h
@@ -0,0 +1,89 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <memory>
+#include <string_view>
+
+#include "storage/index/inverted/analysis/token_batch.h"
+
+namespace doris::segment_v2::inverted_index {
+
+// The tokens of one value at a time, handed over in batches. A consumer keeps
one stream and one
+// batch for many values:
+//
+// stream->reset(value);
+// while (stream->next_batch(batch)) {
+// for (const TokenRef& token : batch.tokens()) { ... }
+// }
+// batch.release_oversized();
+//
+// A stream belongs to one thread; the analyzer that made it may be shared. It
throws only
+// doris::Exception, from reset() and from next_batch(), and is not used again
after it threw.
+class TextTokenStream {
+public:
+ // A streaming source hands over a batch once it holds this many tokens. A
core loop may
+ // overshoot by what one step makes, and a filter that makes several
tokens out of one may
+ // exceed it; a consumer relies only on the return value of next_batch().
+ static constexpr size_t kBatchTokens = 256;
+
+ virtual ~TextTokenStream() = default;
+
+ // Binds `text`, in any state, and drops whatever is left of the previous
value. `text` must
+ // stay valid while the stream iterates over it. An empty `text` makes no
token, whatever its
+ // data pointer: NFKC_CF can fold a value to an empty view that does not
point to null.
+ virtual void reset(std::string_view text) = 0;
+
+ // Clears `batch`, then fills it. Returns true with at least one token, or
false with an empty
+ // batch once the value is exhausted, and only then; it keeps returning
false until the next
+ // reset(). A filter that drops every token of a batch pulls the next one
instead of returning
+ // an empty batch. A term stays valid until the next next_batch() or
reset() on this stream.
+ // An empty term may be in the batch: the consumer decides whether it
counts.
+ virtual bool next_batch(TokenBatch& batch) = 0;
+
+ // The bytes this stream and the streams it wraps keep from one value to
the next. Every buffer
+ // that survives a value counts, and reset() releases those above the high
water mark
+ // (scratch.h), so the number stays bounded however long a value was.
+ virtual size_t reserved_bytes() const = 0;
+
+ // Strict handling of invalid UTF-8, which the pinyin filter needs when it
reports source
+ // offsets: icu, ngram and edge_ngram then throw on it, and ik normalizes
strictly. A filter
+ // forwards it to its input; a tokenizer without a strict form ignores it.
+ virtual void set_strict_utf8(bool /*strict*/) {}
+};
+
+// The start of a chain: makes tokens out of the text.
+class TextTokenizer : public TextTokenStream {};
+
+// A link of a chain: makes tokens out of the tokens of its input, which it
owns.
+class TextTokenFilter : public TextTokenStream {
+public:
+ explicit TextTokenFilter(std::unique_ptr<TextTokenStream> in);
+
+ // Resets the input. A filter with state of its own overrides it and calls
this first.
+ void reset(std::string_view text) override;
+ // The bytes of the input. A filter with buffers of its own overrides it
and adds them.
+ size_t reserved_bytes() const override;
+ void set_strict_utf8(bool strict) override;
+
+protected:
+ std::unique_ptr<TextTokenStream> _in;
+};
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/token_batch.h
b/be/src/storage/index/inverted/analysis/token_batch.h
new file mode 100644
index 00000000000..ebe7fa848b6
--- /dev/null
+++ b/be/src/storage/index/inverted/analysis/token_batch.h
@@ -0,0 +1,102 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <span>
+#include <string_view>
+#include <vector>
+
+#include "common/logging.h"
+#include "core/arena.h"
+#include "storage/index/inverted/analysis/scratch.h"
+
+namespace doris::segment_v2::inverted_index {
+
+// One token: a view of its term, and how many positions it is after the token
before it.
+struct TokenRef {
+ const char* data = nullptr;
+ uint32_t size = 0;
+ int32_t position_increment = 0; // >= 0
+
+ std::string_view term() const { return {data, size}; }
+};
+
+// The tokens a stream hands over in one TextTokenStream::next_batch() call. A
term is a view: of
+// the value being analyzed, of memory the stream keeps, or of the batch's own
term storage, which
+// push_copy() and allocate_term() fill. Term storage is an arena, so growing
it never moves a term
+// handed out before, and clear() drops all of them at once.
+//
+// The consumer owns the batch and keeps it from value to value. Between two
values it calls
+// release_oversized(): the stream cannot, and one huge value must not pin its
memory until the
+// segment ends.
+class TokenBatch {
+public:
+ // Drops the tokens and the terms in term storage. The token array and the
arena's largest
+ // block stay, so the next batch of a steady stream allocates nothing.
+ void clear() {
+ _tokens.clear();
+ _terms.clear();
+ }
+
+ size_t size() const { return _tokens.size(); }
+ std::span<TokenRef> tokens() { return _tokens; }
+ std::span<const TokenRef> tokens() const { return _tokens; }
+
+ // Appends a token whose term stays where it is; whoever owns the term
keeps it valid while the
+ // batch holds the token.
+ void push(std::string_view term, int32_t position_increment) {
+ DCHECK_GE(position_increment, 0);
+ DCHECK_LE(term.size(), std::numeric_limits<uint32_t>::max());
+ _tokens.push_back({.data = term.data(),
+ .size = static_cast<uint32_t>(term.size()),
+ .position_increment = position_increment});
+ }
+
+ // Appends a token with a copy of `term` in term storage.
+ void push_copy(std::string_view term, int32_t position_increment) {
+ char* copy = allocate_term(term.size());
+ term.copy(copy, term.size());
+ push({copy, term.size()}, position_increment);
+ }
+
+ // `size` bytes of term storage, for a term built in place. Valid until
the next clear().
+ char* allocate_term(size_t size) { return _terms.alloc(size); }
+
+ // The bytes the batch keeps across clear(), for the writer's memory
reporter.
+ size_t capacity_bytes() const { return _tokens.capacity() *
sizeof(TokenRef) + _terms.size(); }
+
+ // Between two values, on an empty batch: when the batch keeps more than
the high water mark,
+ // which only a value with a very large batch or very long terms leaves
behind, it releases all
+ // of it. An ordinary batch keeps a 4 KiB token array and a 4 KiB arena
block, and keeps them.
+ void release_oversized() {
+ DCHECK_EQ(_tokens.size(), 0);
+ if (capacity_bytes() > ANALYZER_SCRATCH_HIGH_WATER_BYTES) {
+ std::vector<TokenRef>().swap(_tokens);
+ _terms.clear(/*delete_head=*/true);
+ }
+ }
+
+private:
+ std::vector<TokenRef> _tokens;
+ Arena _terms;
+};
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/analysis/token_stream.h
b/be/src/storage/index/inverted/analysis/token_stream.h
index bdc13c3e9ea..a357523f6e7 100644
--- a/be/src/storage/index/inverted/analysis/token_stream.h
+++ b/be/src/storage/index/inverted/analysis/token_stream.h
@@ -31,6 +31,7 @@
#include "CLucene/analysis/AnalysisHeader.h"
#include "common/cast_set.h"
#include "storage/index/inverted/analysis/reader.h"
+#include "storage/index/inverted/analysis/scratch.h"
using namespace lucene::analysis;
@@ -41,10 +42,6 @@ using TokenizerPtr = std::shared_ptr<DorisTokenizer>;
using TokenStreamPtr = std::shared_ptr<TokenStream>;
-// Reused analyzers keep ordinary scratch across values; one oversized value
must not pin its
-// capacity for the rest of the writer lifetime.
-constexpr size_t ANALYZER_SCRATCH_HIGH_WATER_BYTES = 64 * 1024;
-
// Longest prefix of text that fits in max_bytes without splitting a rune, and
its rune count.
inline std::pair<size_t, size_t> utf8_prefix_at_most(std::string_view text,
size_t max_bytes) {
const auto length = static_cast<int32_t>(text.size());
@@ -63,14 +60,6 @@ inline std::pair<size_t, size_t>
utf8_prefix_at_most(std::string_view text, size
return {static_cast<size_t>(offset), rune_count};
}
-template <typename Container>
-void release_oversized_scratch(Container& container) {
- if (container.capacity() * sizeof(typename Container::value_type) >
- ANALYZER_SCRATCH_HIGH_WATER_BYTES) {
- Container().swap(container);
- }
-}
-
/**
* All custom tokenizers and token_filters must use the following functions
* to set token information. Using these unified set methods helps avoid
diff --git a/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.cpp
b/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.cpp
new file mode 100644
index 00000000000..cab1d251e89
--- /dev/null
+++ b/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.cpp
@@ -0,0 +1,111 @@
+// 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 "storage/index/inverted/clucene/lucene_analyzer_bridge.h"
+
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/check.h"
+#include "common/exception.h"
+
+namespace doris::segment_v2::inverted_index {
+
+namespace {
+
+// The whole value of `reader`, without a copy. SStringReader::read() hands
out a view of its bytes
+// and returns -1, not 0, for an empty value.
+std::string_view read_value(const ReaderPtr& reader) {
+ const char* start = nullptr;
+ const int32_t length = reader->read(reinterpret_cast<const
void**>(&start), 0,
+ cast_set<int32_t>(reader->size()));
+ return length > 0 ? std::string_view(start, length) : std::string_view();
+}
+
+} // namespace
+
+LuceneTokenStreamBridge::LuceneTokenStreamBridge(std::unique_ptr<TextTokenStream>
stream)
+ : _stream(std::move(stream)) {
+ DORIS_CHECK(_stream != nullptr);
+}
+
+LuceneTokenStreamBridge::~LuceneTokenStreamBridge() = default;
+
+void LuceneTokenStreamBridge::bind(std::string_view text) {
+ _pending = text;
+}
+
+void LuceneTokenStreamBridge::reset() {
+ // Exhausted until the stream accepted the value, so a value whose reset()
threw makes no token.
+ _exhausted = true;
+ _batch.clear();
+ _cursor = 0;
+ _stream->reset(_pending);
+ _exhausted = false;
+}
+
+lucene::analysis::Token*
LuceneTokenStreamBridge::next(lucene::analysis::Token* token) {
+ if (_cursor == _batch.size()) {
+ _cursor = 0;
+ if (_exhausted || !_stream->next_batch(_batch)) {
+ _exhausted = true;
+ _batch.release_oversized();
+ return nullptr;
+ }
+ }
+ const TokenRef& ref = _batch.tokens()[_cursor++];
+ token->setTextNoCopy(ref.data, static_cast<int32_t>(ref.size));
+ token->setPositionIncrement(ref.position_increment);
+ return token;
+}
+
+LuceneAnalyzerBridge::LuceneAnalyzerBridge(TextAnalyzerPtr analyzer)
+ : _analyzer(std::move(analyzer)) {
+ DORIS_CHECK(_analyzer != nullptr);
+}
+
+LuceneAnalyzerBridge::~LuceneAnalyzerBridge() = default;
+
+lucene::analysis::TokenStream* LuceneAnalyzerBridge::tokenStream(const TCHAR*
/*field_name*/,
+ const
ReaderPtr& reader) {
+ auto stream =
std::make_unique<LuceneTokenStreamBridge>(_analyzer->create_stream());
+ stream->bind(read_value(reader));
+ return stream.release();
+}
+
+lucene::analysis::TokenStream* LuceneAnalyzerBridge::reusableTokenStream(
+ const TCHAR* /*field_name*/, const ReaderPtr& reader) {
+ if (_reusable == nullptr) {
+ _reusable =
std::make_unique<LuceneTokenStreamBridge>(_analyzer->create_stream());
+ }
+ _reusable->bind(read_value(reader));
+ return _reusable.get();
+}
+
+lucene::analysis::TokenStream* LuceneAnalyzerBridge::tokenStream(const TCHAR*
/*field_name*/,
+
lucene::util::Reader* /*reader*/) {
+ throw Exception(ErrorCode::INVERTED_INDEX_NOT_SUPPORTED,
+ "LuceneAnalyzerBridge::tokenStream needs a ReaderPtr");
+}
+
+lucene::analysis::TokenStream* LuceneAnalyzerBridge::reusableTokenStream(
+ const TCHAR* /*field_name*/, lucene::util::Reader* /*reader*/) {
+ throw Exception(ErrorCode::INVERTED_INDEX_NOT_SUPPORTED,
+ "LuceneAnalyzerBridge::reusableTokenStream needs a
ReaderPtr");
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.h
b/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.h
new file mode 100644
index 00000000000..589f52df51a
--- /dev/null
+++ b/be/src/storage/index/inverted/clucene/lucene_analyzer_bridge.h
@@ -0,0 +1,88 @@
+// 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.
+
+#pragma once
+
+#include <CLucene.h> // IWYU pragma: keep
+
+#include <cstddef>
+#include <memory>
+#include <string_view>
+
+#include "CLucene/analysis/AnalysisHeader.h"
+#include "storage/index/inverted/analysis/reader.h"
+#include "storage/index/inverted/analysis/text_analyzer.h"
+#include "storage/index/inverted/analysis/token_batch.h"
+
+namespace doris::segment_v2::inverted_index {
+
+// A TextTokenStream behind the CLucene TokenStream interface, for the V1 and
V2 writers. Terms are
+// handed over without a copy: a term stays valid until the next batch, and
CLucene is done with a
+// token before it asks for the next one. An empty term is handed over too,
because the V1 and V2
+// writers give it a position. A doris::Exception passes through the CLucene
frames to the caller
+// of IndexWriter::addDocument(), whose _CLFINALLY blocks clean up on the way.
+class LuceneTokenStreamBridge final : public lucene::analysis::TokenStream {
+public:
+ explicit LuceneTokenStreamBridge(std::unique_ptr<TextTokenStream> stream);
+ ~LuceneTokenStreamBridge() override;
+
+ // Parks the value; reset() starts it.
+ void bind(std::string_view text);
+ // SDocumentsWriter::invertField() calls reset() once per field, and that
is what starts the
+ // parked value, exactly as DorisTokenizer::reset() promotes its pending
reader today.
+ void reset() override;
+ lucene::analysis::Token* next(lucene::analysis::Token* token) override;
+ // invertField() closes the stream after every field, and the scalar path
reuses it.
+ void close() override {}
+
+private:
+ std::unique_ptr<TextTokenStream> _stream;
+ TokenBatch _batch;
+ size_t _cursor = 0; // the next token of _batch
+ std::string_view _pending;
+ // No value is started, or the stream is done with it.
+ bool _exhausted = true;
+};
+
+// A TextAnalyzer behind the CLucene Analyzer interface, for the V1 and V2
writers. The writer
+// passes each value as an SStringReader on the column's bytes; the char
filters are in the
+// TextAnalyzer.
+class LuceneAnalyzerBridge final : public lucene::analysis::Analyzer {
+public:
+ explicit LuceneAnalyzerBridge(TextAnalyzerPtr analyzer);
+ ~LuceneAnalyzerBridge() override;
+
+ bool isSDocOpt() override { return true; }
+
+ // A new stream the caller owns. Array elements take this path, one stream
per element.
+ lucene::analysis::TokenStream* tokenStream(const TCHAR* field_name,
+ const ReaderPtr& reader)
override;
+ // The bridge's own stream, rebound to the value. A scalar column takes
this path.
+ lucene::analysis::TokenStream* reusableTokenStream(const TCHAR* field_name,
+ const ReaderPtr&
reader) override;
+ // Pure virtual in the base and never reached, as in CustomAnalyzer: these
throw.
+ lucene::analysis::TokenStream* tokenStream(const TCHAR* field_name,
+ lucene::util::Reader* reader)
override;
+ lucene::analysis::TokenStream* reusableTokenStream(const TCHAR* field_name,
+ lucene::util::Reader*
reader) override;
+
+private:
+ const TextAnalyzerPtr _analyzer;
+ std::unique_ptr<LuceneTokenStreamBridge> _reusable;
+};
+
+} // namespace doris::segment_v2::inverted_index
diff --git
a/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.cpp
b/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.cpp
new file mode 100644
index 00000000000..61965848ef6
--- /dev/null
+++ b/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.cpp
@@ -0,0 +1,117 @@
+// 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 "storage/index/inverted/clucene/lucene_backed_text_analyzer.h"
+
+#include <CLucene.h>
+
+#include <string_view>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/check.h"
+#include "common/exception.h"
+#include "storage/index/inverted/analysis/analyzer/analyzer.h"
+#include "storage/index/inverted/analysis/reader.h"
+
+namespace doris::segment_v2::inverted_index {
+
+namespace {
+
+// One value at a time through a CLucene analyzer. The reader, the token and
the CLucene stream are
+// made by reset() and released once the value is exhausted, so nothing
outlives a value.
+class LuceneBackedTokenStream final : public TextTokenizer {
+public:
+ // Both outlive the stream, because a stream does not outlive its analyzer.
+ LuceneBackedTokenStream(lucene::analysis::Analyzer* analyzer, const
CharFilterMap* char_filters)
+ : _analyzer(analyzer), _char_filters(char_filters) {}
+
+ void reset(std::string_view text) override {
+ release();
+ try {
+ _reader = InvertedIndexAnalyzer::create_reader(*_char_filters);
+ _reader->init(text.data(), cast_set<int32_t>(text.size()), false);
+ _token = std::make_unique<lucene::analysis::Token>();
+ _stream.reset(_analyzer->tokenStream(L"", _reader));
+ } catch (const CLuceneError& e) {
+ release();
+ throw Exception(ErrorCode::INVERTED_INDEX_ANALYZER_ERROR,
+ "the CLucene analyzer failed to start a value:
{}", e.what());
+ }
+ }
+
+ bool next_batch(TokenBatch& batch) override {
+ batch.clear();
+ while (_stream != nullptr && batch.size() < kBatchTokens) {
+ lucene::analysis::Token* token = next_lucene_token();
+ if (token == nullptr) {
+ release();
+ break;
+ }
+ batch.push_copy({token->termBuffer<char>(),
token->termLength<char>()},
+ token->getPositionIncrement());
+ }
+ return batch.size() != 0;
+ }
+
+ // Nothing survives a value.
+ size_t reserved_bytes() const override { return 0; }
+
+private:
+ // The next CLucene token, or nullptr once the value is exhausted.
+ lucene::analysis::Token* next_lucene_token() {
+ try {
+ lucene::analysis::Token* token = _stream->next(_token.get());
+ if (token == nullptr) {
+ _stream->close();
+ }
+ return token;
+ } catch (const CLuceneError& e) {
+ release();
+ throw Exception(ErrorCode::INVERTED_INDEX_ANALYZER_ERROR,
+ "the CLucene analyzer failed on a value: {}",
e.what());
+ }
+ }
+
+ void release() {
+ _stream.reset();
+ _token.reset();
+ _reader.reset();
+ }
+
+ lucene::analysis::Analyzer* const _analyzer;
+ const CharFilterMap* const _char_filters;
+ ReaderPtr _reader;
+ std::unique_ptr<lucene::analysis::Token> _token;
+ std::unique_ptr<lucene::analysis::TokenStream> _stream;
+};
+
+} // namespace
+
+LuceneBackedTextAnalyzer::LuceneBackedTextAnalyzer(
+ std::shared_ptr<lucene::analysis::Analyzer> analyzer, CharFilterMap
char_filters)
+ : _analyzer(std::move(analyzer)),
_char_filters(std::move(char_filters)) {
+ DORIS_CHECK(_analyzer != nullptr);
+}
+
+LuceneBackedTextAnalyzer::~LuceneBackedTextAnalyzer() = default;
+
+std::unique_ptr<TextTokenStream> LuceneBackedTextAnalyzer::create_stream()
const {
+ return std::make_unique<LuceneBackedTokenStream>(_analyzer.get(),
&_char_filters);
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git
a/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.h
b/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.h
new file mode 100644
index 00000000000..16610072867
--- /dev/null
+++ b/be/src/storage/index/inverted/clucene/lucene_backed_text_analyzer.h
@@ -0,0 +1,53 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+
+#include "storage/index/inverted/analysis/inverted_index_parser.h"
+#include "storage/index/inverted/analysis/text_analyzer.h"
+
+namespace lucene::analysis {
+class Analyzer;
+} // namespace lucene::analysis
+
+namespace doris::segment_v2::inverted_index {
+
+// A CLucene analyzer behind the TextAnalyzer interface, for the consumers
that move to the
+// interface before the analyzers do. It goes once every analyzer is native.
+//
+// Every value gets a new reader through the index's char filter
(InvertedIndexAnalyzer::
+// create_reader) and a new CLucene stream from Analyzer::tokenStream(), as a
query reads a value
+// today; the SNII writer does the same with one reader it re-initializes.
Sharing the analyzer
+// between threads is as safe as sharing the CLucene analyzer for
tokenStream() calls. Terms are
+// copied into the batch, because a CLucene token is overwritten by the next
one; a CLuceneError
+// comes out as a doris::Exception.
+class LuceneBackedTextAnalyzer final : public TextAnalyzer {
+public:
+ LuceneBackedTextAnalyzer(std::shared_ptr<lucene::analysis::Analyzer>
analyzer,
+ CharFilterMap char_filters);
+ ~LuceneBackedTextAnalyzer() override;
+
+ std::unique_ptr<TextTokenStream> create_stream() const override;
+
+private:
+ const std::shared_ptr<lucene::analysis::Analyzer> _analyzer;
+ const CharFilterMap _char_filters;
+};
+
+} // namespace doris::segment_v2::inverted_index
diff --git
a/be/test/storage/index/inverted/analysis/golden/analysis_golden_test.cpp
b/be/test/storage/index/inverted/analysis/golden/analysis_golden_test.cpp
index 6221a4c7ff6..9521f4204ef 100644
--- a/be/test/storage/index/inverted/analysis/golden/analysis_golden_test.cpp
+++ b/be/test/storage/index/inverted/analysis/golden/analysis_golden_test.cpp
@@ -48,6 +48,7 @@
#include <set>
#include <sstream>
#include <string>
+#include <thread>
#include <utility>
#include <vector>
@@ -99,6 +100,9 @@ void write_dump(const std::string& dir, const std::string&
key, DriveMode mode,
out << dump;
}
+// Makes the driver of a configuration: make_golden_driver(), or
make_golden_text_driver().
+using DriverFactory = std::function<Status(const GoldenConfig&,
std::unique_ptr<GoldenDriver>*)>;
+
// One configuration over one family in one mode.
struct GoldenJob {
const GoldenConfig* config = nullptr;
@@ -111,11 +115,11 @@ struct GoldenJob {
double seconds = 0;
};
-void run_job(GoldenJob* job, bool dump) {
+void run_job(GoldenJob* job, bool dump, const DriverFactory& make_driver) {
const auto started = std::chrono::steady_clock::now();
// A driver per job, so no family sees the state another one left.
std::unique_ptr<GoldenDriver> driver;
- if (const Status status = make_golden_driver(*job->config, &driver);
!status.ok()) {
+ if (const Status status = make_driver(*job->config, &driver);
!status.ok()) {
job->error = status.to_string();
} else {
try {
@@ -222,31 +226,74 @@ void print_timing(const std::vector<GoldenJob>& jobs,
double wall) {
}
}
-// Runs `configs` over `families` in `requested_modes`, and in the mode whose
output is frozen, and
-// checks every digest against `file`. The jobs run on several threads; the
checks run here, in
-// the order of the matrix.
-void check_golden(const std::vector<GoldenConfig>& configs,
- const std::vector<CorpusFamily>& families,
- const std::vector<DriveMode>& requested_modes,
test::GoldenFile* file) {
+// The configurations ANALYSIS_GOLDEN_ONLY selects: those whose name contains
it, or all.
+std::vector<const GoldenConfig*> selected_configs(const
std::vector<GoldenConfig>& configs) {
const std::string only = env("ANALYSIS_GOLDEN_ONLY");
- const std::string dump_dir = env("ANALYSIS_GOLDEN_DUMP");
std::vector<const GoldenConfig*> selected;
for (const GoldenConfig& config : configs) {
if (only.empty() || config.name.find(only) != std::string::npos) {
selected.push_back(&config);
}
}
+ return selected;
+}
+
+// The families of `families` that `names` names, in their order.
+std::vector<CorpusFamily> families_named(std::vector<CorpusFamily> families,
+ const std::set<std::string>& names) {
+ std::erase_if(families,
+ [&](const CorpusFamily& family) { return
!names.contains(family.name); });
+ return families;
+}
+
+struct ThreadDigest {
+ FamilyDigest digest;
+ std::string error;
+};
+
+// The digests of `family` from `threads` threads that share `analyzer`, the
even ones with a new
+// stream per line and the odd ones with one stream. A thread that failed has
its error instead.
+std::vector<ThreadDigest> digests_on_threads(const TextAnalyzerPtr& analyzer,
+ const CorpusFamily& family,
size_t threads) {
+ std::vector<ThreadDigest> results(threads);
+ std::vector<std::thread> workers;
+ for (size_t t = 0; t < threads; ++t) {
+ workers.emplace_back([&, t] {
+ try {
+ const auto driver = make_text_driver(analyzer);
+ results[t].digest = digest_family(
+ *driver, t % 2 == 0 ? DriveMode::FRESH :
DriveMode::REUSED, family);
+ } catch (const std::exception& e) {
+ results[t].error = e.what();
+ }
+ });
+ }
+ for (std::thread& worker : workers) {
+ worker.join();
+ }
+ return results;
+}
+
+// Runs `configs` over `families` in `requested_modes`, and in the mode whose
output is frozen, and
+// checks every digest against `file`. The jobs run on several threads; the
checks run here, in
+// the order of the matrix.
+void check_golden(const std::vector<GoldenConfig>& configs,
+ const std::vector<CorpusFamily>& families,
+ const std::vector<DriveMode>& requested_modes,
test::GoldenFile* file,
+ const DriverFactory& make_driver = make_golden_driver) {
+ const std::string dump_dir = env("ANALYSIS_GOLDEN_DUMP");
+ const std::vector<const GoldenConfig*> selected =
selected_configs(configs);
// Build each configuration and analyze one line before going parallel, so
a process-wide
// dictionary is loaded by one thread, as in a BE that has served a first
query.
for (const GoldenConfig* config : selected) {
std::unique_ptr<GoldenDriver> driver;
- const Status status = make_golden_driver(*config, &driver);
+ const Status status = make_driver(*config, &driver);
ASSERT_TRUE(status.ok()) << status.to_string() << "\n" <<
describe(*config);
driver->fresh("warm up 预热");
}
std::vector<GoldenJob> jobs = make_jobs(selected, families,
requested_modes);
const auto started = std::chrono::steady_clock::now();
- parallel_for(jobs.size(), [&](size_t i) { run_job(&jobs[i],
!dump_dir.empty()); });
+ parallel_for(jobs.size(), [&](size_t i) { run_job(&jobs[i],
!dump_dir.empty(), make_driver); });
const double wall =
std::chrono::duration<double>(std::chrono::steady_clock::now() -
started).count();
for (size_t i = 0; i < jobs.size();) {
@@ -266,6 +313,15 @@ protected:
ASSERT_TRUE(status.ok()) << status.to_string();
}
+ // The configurations that build an analyzer.
+ std::vector<GoldenConfig> analyzer_configs() const {
+ std::vector<GoldenConfig> configs;
+ std::ranges::copy_if(_configs, std::back_inserter(configs), [](const
GoldenConfig& config) {
+ return config.kind == GoldenKind::ANALYZE;
+ });
+ return configs;
+ }
+
const std::vector<GoldenConfig> _configs = golden_configs();
GoldenEnvironment _environment;
};
@@ -277,6 +333,45 @@ TEST_F(AnalysisGolden, MatchesTheFrozenDigests) {
file.finish(owns_analysis_line, env("ANALYSIS_GOLDEN_ONLY").empty());
}
+// The consumers move to the TextAnalyzer interface before the analyzers do,
through
+// LuceneBackedTextAnalyzer. Through it every analyzer gives the frozen
output, with a new stream
+// per line and with one stream for all lines.
+TEST_F(AnalysisGolden, LuceneBackedTextAnalyzerMatchesTheFrozenDigests) {
+ test::GoldenFile file(kGoldenFile);
+ check_golden(analyzer_configs(),
default_families(_environment.corpus_sources()),
+ {DriveMode::FRESH, DriveMode::REUSED}, &file,
make_golden_text_driver);
+ file.finish(owns_analysis_line, false);
+}
+
+// An analyzer serves many threads at once. Eight threads share the
LuceneBackedTextAnalyzer of
+// each configuration, half of them with a new stream per line and half with
one stream, and each
+// thread gives the frozen output.
+TEST_F(AnalysisGolden, EightThreadsShareOneTextAnalyzer) {
+ const std::vector<CorpusFamily> families = families_named(
+ default_families(_environment.corpus_sources()), {"mixed_zh_en",
"invalid_utf8"});
+ ASSERT_EQ(families.size(), 2);
+ const std::vector<GoldenConfig> configs = analyzer_configs();
+ test::GoldenFile file(kGoldenFile);
+ for (const GoldenConfig* config : selected_configs(configs)) {
+ TextAnalyzerPtr analyzer;
+ const Status status = make_golden_text_analyzer(*config, &analyzer);
+ ASSERT_TRUE(status.ok()) << status.to_string() << "\n" <<
describe(*config);
+ for (const CorpusFamily& family : families) {
+ const std::vector<ThreadDigest> results =
digests_on_threads(analyzer, family, 8);
+ for (size_t t = 0; t < results.size(); ++t) {
+ ASSERT_TRUE(results[t].error.empty())
+ << config->name << " thread " << t << ": " <<
results[t].error;
+ EXPECT_EQ(results[t].digest, results[0].digest)
+ << config->name << " " << family.name << ": thread "
<< t
+ << " disagrees with thread 0";
+ }
+ file.check(config->name, family.name, results[0].digest.tokens,
+ results[0].digest.digest, describe(*config));
+ }
+ }
+ file.finish(owns_analysis_line, false);
+}
+
// Every component the analysis factory registers is in some configuration, so
a new component
// cannot ship without frozen output.
TEST_F(AnalysisGolden, EveryRegisteredComponentHasAConfiguration) {
@@ -332,12 +427,8 @@ TEST_F(AnalysisGolden, IcuIgnoresTheDefaultLocale) {
return config.name.find("icu") != std::string::npos;
});
ASSERT_FALSE(icu_configs.empty());
- std::vector<CorpusFamily> families;
- for (CorpusFamily& family :
default_families(_environment.corpus_sources())) {
- if (family.name == "latin_words" || family.name == "bmp_code_points") {
- families.push_back(std::move(family));
- }
- }
+ const std::vector<CorpusFamily> families = families_named(
+ default_families(_environment.corpus_sources()), {"latin_words",
"bmp_code_points"});
ASSERT_EQ(families.size(), 2);
test::GoldenFile file(kGoldenFile);
check_golden(icu_configs, families, {DriveMode::FRESH}, &file);
diff --git
a/be/test/storage/index/inverted/analysis/golden/analysis_write_golden_test.cpp
b/be/test/storage/index/inverted/analysis/golden/analysis_write_golden_test.cpp
index 9ab4309b19e..38d37d16bc2 100644
---
a/be/test/storage/index/inverted/analysis/golden/analysis_write_golden_test.cpp
+++
b/be/test/storage/index/inverted/analysis/golden/analysis_write_golden_test.cpp
@@ -27,8 +27,11 @@
#include <gtest/gtest.h>
#include <algorithm>
+#include <climits>
#include <fstream>
+#include <functional>
#include <memory>
+#include <set>
#include <sstream>
#include <string>
#include <vector>
@@ -42,6 +45,10 @@
#include "storage/index/inverted/analysis/golden/golden_configs.h"
#include "storage/index/inverted/analysis/golden/golden_corpus.h"
#include "storage/index/inverted/analysis/golden/golden_environment.h"
+#include "storage/index/inverted/analysis/inverted_index_parser.h"
+#include "storage/index/inverted/clucene/inverted_index_writer.h"
+#include "storage/index/inverted/clucene/lucene_analyzer_bridge.h"
+#include "storage/index/inverted/clucene/lucene_backed_text_analyzer.h"
#include "storage/index/inverted/common/inverted_index_desc.h"
#include "storage/tablet/tablet_schema.h"
#include "testutil/golden_file.h"
@@ -66,6 +73,17 @@ constexpr const char* kConfigs[] = {
"custom/char_replace+standard+word_delimiter+asciifolding+lowercase",
};
+// The configurations of kConfigs whose array elements the V1 and V2 writers
index nothing of. The
+// writer asks Analyzer::tokenStream() for a stream per element, ICUAnalyzer
resets that stream
+// before it returns it, and SDocumentsWriter::invertField() resets it again.
ICUTokenizer reads its
+// reader in reset(), so the second time it finds the reader consumed and
makes no token: an
+// element "hello world" is indexed as nothing. LuceneAnalyzerBridge starts
each element once, so
+// through it the elements are indexed. No table has such an index: the FE
accepts no parser and no
+// analyzer on an array column (InvertedIndexUtil.checkInvertedIndexParser).
+const std::set<std::string> kOldArrayPathLosesElements = {
+ "builtin/icu/lower_case=unset/stopwords=unset",
+};
+
struct Format {
const char* name;
InvertedIndexStorageFormatPB format;
@@ -115,6 +133,24 @@ TabletColumn text_column() {
return column;
}
+TabletColumn array_column() {
+ TabletColumn element;
+ element.set_name("text");
+ element.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
+ element.set_length(INT_MAX);
+ element.set_is_nullable(true);
+ TabletColumn column;
+ column.set_unique_id(0);
+ column.set_name("texts");
+ column.set_type(FieldType::OLAP_FIELD_TYPE_ARRAY);
+ column.set_is_nullable(true);
+ column.add_sub_column(element);
+ return column;
+}
+
+// Called on the column writer before the rows go in.
+using WriterHook = std::function<void(IndexColumnWriter*)>;
+
// What every index gets: the first lines of four families, one batch each,
nulls after each.
std::vector<std::vector<std::string>> row_batches(const
std::vector<CorpusFamily>& families) {
std::vector<std::vector<std::string>> batches;
@@ -131,7 +167,8 @@ std::vector<std::vector<std::string>> row_batches(const
std::vector<CorpusFamily
// Writes the batches through the production writer factory; returns the index
path prefix.
std::string write_index(const std::string& name, InvertedIndexStorageFormatPB
format,
const TabletIndex& meta,
- const std::vector<std::vector<std::string>>& batches) {
+ const std::vector<std::vector<std::string>>& batches,
+ const WriterHook& before_rows = {}) {
const std::string prefix = std::string(kTestDir) + "/" + name;
const auto fs = io::global_local_filesystem();
io::FileWriterPtr file_writer;
@@ -150,6 +187,9 @@ std::string write_index(const std::string& name,
InvertedIndexStorageFormatPB fo
if (!created.ok()) {
return prefix;
}
+ if (before_rows) {
+ before_rows(writer.get());
+ }
for (const std::vector<std::string>& batch : batches) {
std::vector<Slice> slices;
for (const std::string& row : batch) {
@@ -166,6 +206,74 @@ std::string write_index(const std::string& name,
InvertedIndexStorageFormatPB fo
return prefix;
}
+// The lines of the batches as array rows of 0 to 3 elements, with a null
element now and then.
+struct ArrayRows {
+ std::vector<Slice> elements;
+ std::vector<uint8_t> null_map;
+ std::vector<uint64_t> offsets; // one more than the rows
+};
+
+ArrayRows array_rows(const std::vector<std::vector<std::string>>& batches) {
+ ArrayRows rows;
+ rows.offsets.push_back(0);
+ for (const std::vector<std::string>& batch : batches) {
+ for (size_t i = 0; i < batch.size();) {
+ const size_t size = std::min((rows.offsets.size() - 1) % 4,
batch.size() - i);
+ for (size_t end = i + size; i < end; ++i) {
+ rows.null_map.push_back(rows.elements.size() % 7 == 3 ? 1 : 0);
+ rows.elements.emplace_back(batch[i]);
+ }
+ rows.offsets.push_back(rows.elements.size());
+ }
+ }
+ return rows;
+}
+
+// Writes the rows into the V2 index of an array column; returns the index
path prefix.
+std::string write_array_index(const std::string& name, const TabletIndex& meta,
+ const ArrayRows& rows, const WriterHook&
before_rows = {}) {
+ const std::string prefix = std::string(kTestDir) + "/" + name;
+ const auto fs = io::global_local_filesystem();
+ io::FileWriterPtr file_writer;
+ EXPECT_TRUE(
+
fs->create_file(InvertedIndexDescriptor::get_index_file_path_v2(prefix),
&file_writer)
+ .ok());
+ IndexFileWriter index_file_writer(fs, prefix, "write_golden_rowset",
/*seg_id=*/0,
+ InvertedIndexStorageFormatPB::V2,
std::move(file_writer),
+ /*can_use_ram_dir=*/true,
/*tablet_id=*/7700);
+ const TabletColumn column = array_column();
+ std::unique_ptr<IndexColumnWriter> writer;
+ const Status created = IndexColumnWriter::create(&column, &writer,
&index_file_writer, &meta);
+ EXPECT_TRUE(created.ok()) << created.to_string();
+ if (!created.ok()) {
+ return prefix;
+ }
+ if (before_rows) {
+ before_rows(writer.get());
+ }
+ const Status added = writer->add_array_values(
+ sizeof(Slice), reinterpret_cast<const
uint8_t*>(rows.elements.data()),
+ rows.null_map.data(), rows.offsets.data(), rows.offsets.size() -
1);
+ EXPECT_TRUE(added.ok()) << added.to_string();
+ EXPECT_TRUE(writer->finish().ok());
+ EXPECT_TRUE(index_file_writer.begin_close().ok());
+ EXPECT_TRUE(index_file_writer.finish_close().ok());
+ return prefix;
+}
+
+// Puts the V2 writer's analyzer behind LuceneBackedTextAnalyzer and
LuceneAnalyzerBridge, as the
+// writer sees it once the consumers use the TextAnalyzer interface; the
bridge reads each value
+// from a plain reader. The CLucene IndexWriter keeps a raw pointer to the
analyzer it was made
+// with, which the LuceneBackedTextAnalyzer keeps alive. The index has no char
filter of its own:
+// the writer gives array elements the index's char filter itself, which would
apply it twice.
+void use_bridge(IndexColumnWriter* writer) {
+ auto* legacy =
+
static_cast<InvertedIndexColumnWriter<FieldType::OLAP_FIELD_TYPE_STRING>*>(writer);
+ auto analyzer =
std::make_shared<LuceneBackedTextAnalyzer>(legacy->_analyzer, CharFilterMap {});
+
legacy->set_analysis_for_test(std::make_shared<lucene::util::SStringReader<char>>(),
+
std::make_shared<LuceneAnalyzerBridge>(std::move(analyzer)));
+}
+
struct WriteDigest {
uint64_t count = 0;
uint64_t digest = 0;
@@ -283,4 +391,44 @@ TEST_F(AnalysisWriteGolden, MatchesTheFrozenDigests) {
file.finish([](const std::string& name) { return
name.starts_with(kWritePrefix); }, true);
}
+// The V1 and V2 writers reach the analyzers through LuceneAnalyzerBridge once
the consumers use
+// the TextAnalyzer interface. Through the bridge, over the analyzer in a
LuceneBackedTextAnalyzer,
+// the V2 writer writes the terms, documents and positions it writes with the
analyzer itself: the
+// frozen ones for a string column, and for an array column, whose elements
take the other way
+// into the analyzer, the ones the analyzer itself gives, except where it
gives none
+// (kOldArrayPathLosesElements).
+TEST_F(AnalysisWriteGolden, LuceneAnalyzerBridgeWritesTheSameV2Index) {
+ const std::vector<std::vector<std::string>> batches =
+ row_batches(default_families(_environment.corpus_sources()));
+ const ArrayRows arrays = array_rows(batches);
+ test::GoldenFile file(kGoldenFile);
+ for (const char* name : kConfigs) {
+ const auto config = std::ranges::find_if(
+ _configs, [&](const GoldenConfig& c) { return c.name == name;
});
+ ASSERT_NE(config, _configs.end()) << name;
+
ASSERT_TRUE(get_parser_char_filter_map_from_properties(config->properties).empty())
<< name;
+ const TabletIndex meta = index_meta(*config);
+ const std::string id = std::to_string(config - _configs.begin());
+ constexpr auto kV2 = InvertedIndexStorageFormatPB::V2;
+
+ const WriteDigest scalar = clucene_digest(
+ write_index("bridge_" + id, kV2, meta, batches, use_bridge),
kV2, meta);
+ file.check(std::string(kWritePrefix) + "v2/" + config->name, "rows",
scalar.count,
+ scalar.digest, describe(*config));
+
+ const WriteDigest array =
+ clucene_digest(write_array_index("array_" + id, meta, arrays),
kV2, meta);
+ const WriteDigest bridged_array = clucene_digest(
+ write_array_index("array_bridge_" + id, meta, arrays,
use_bridge), kV2, meta);
+ EXPECT_GT(bridged_array.count, 0) << name;
+ if (kOldArrayPathLosesElements.contains(name)) {
+ EXPECT_EQ(array.count, 0) << name;
+ } else {
+ EXPECT_EQ(bridged_array.count, array.count) << name;
+ EXPECT_EQ(bridged_array.digest, array.digest) << name;
+ }
+ }
+ file.finish([](const std::string& name) { return
name.starts_with(kWritePrefix); }, false);
+}
+
} // namespace doris::segment_v2::inverted_index::golden
diff --git a/be/test/storage/index/inverted/analysis/golden/golden_driver.cpp
b/be/test/storage/index/inverted/analysis/golden/golden_driver.cpp
index b3720240d01..f58a7facade 100644
--- a/be/test/storage/index/inverted/analysis/golden/golden_driver.cpp
+++ b/be/test/storage/index/inverted/analysis/golden/golden_driver.cpp
@@ -37,8 +37,10 @@
#include "storage/index/inverted/analysis/char_filter/char_filter_factory.h"
#include "storage/index/inverted/analysis/inverted_index_parser.h"
#include "storage/index/inverted/analysis/reader.h"
+#include "storage/index/inverted/analysis/token_batch.h"
#include "storage/index/inverted/analysis/token_filter/token_filter_factory.h"
#include "storage/index/inverted/analysis/tokenizer/tokenizer_factory.h"
+#include "storage/index/inverted/clucene/lucene_backed_text_analyzer.h"
#include "util/hash_util.hpp"
namespace doris::segment_v2::inverted_index::golden {
@@ -182,6 +184,66 @@ private:
const std::map<std::string, std::string> _properties;
};
+class TextDriver final : public GoldenDriver {
+public:
+ explicit TextDriver(TextAnalyzerPtr analyzer) :
_analyzer(std::move(analyzer)) {}
+
+ GoldenLine fresh(std::string_view line) override {
+ std::unique_ptr<TextTokenStream> stream;
+ return read(&stream, line);
+ }
+
+ GoldenLine reused(std::string_view line) override { return read(&_stream,
line); }
+
+private:
+ // Reads `line` with `*stream`, made first if there is none. A stream is
not used again after
+ // it threw, so it is dropped then.
+ GoldenLine read(std::unique_ptr<TextTokenStream>* stream, std::string_view
line) {
+ GoldenLine out;
+ try {
+ if (*stream == nullptr) {
+ *stream = _analyzer->create_stream();
+ }
+ (*stream)->reset(line);
+ while ((*stream)->next_batch(_batch)) {
+ for (const TokenRef& token : _batch.tokens()) {
+ out.tokens.push_back({.bytes = std::string(token.term()),
+ .position_increment =
token.position_increment});
+ }
+ }
+ } catch (const Exception&) {
+ out = threw_line();
+ stream->reset();
+ _batch.clear();
+ }
+ _batch.release_oversized();
+ return out;
+ }
+
+ const TextAnalyzerPtr _analyzer;
+ std::unique_ptr<TextTokenStream> _stream;
+ TokenBatch _batch;
+};
+
+// The analyzer configuration the writers read from `properties`, except that
an unset lower_case
+// stays unset, as on the query path; lower_case=true covers what the writers
pass.
+InvertedIndexAnalyzerConfig analyzer_config_of(
+ const std::map<std::string, std::string>& properties) {
+ InvertedIndexAnalyzerConfig analyzer_config;
+ analyzer_config.analyzer_name =
get_analyzer_name_from_properties(properties);
+ analyzer_config.parser_type = get_inverted_index_parser_type_from_string(
+ get_parser_string_from_properties(properties));
+ analyzer_config.parser_mode =
get_parser_mode_string_from_properties(properties);
+ analyzer_config.lower_case =
get_parser_lowercase_from_properties(properties);
+ analyzer_config.stop_words =
get_parser_stopwords_from_properties(properties);
+ analyzer_config.char_filter_map =
get_parser_char_filter_map_from_properties(properties);
+ return analyzer_config;
+}
+
+AnalyzerPtr build_analyzer(const InvertedIndexAnalyzerConfig& analyzer_config)
{
+ return
InvertedIndexAnalyzer::create_analyzer_provider(&analyzer_config)->get_analyzer();
+}
+
} // namespace
FamilyDigest digest_family(GoldenDriver& driver, DriveMode mode, const
CorpusFamily& family,
@@ -260,23 +322,10 @@ Status make_golden_driver(const GoldenConfig& config,
std::unique_ptr<GoldenDriv
*driver = std::make_unique<NormalizeDriver>(config.properties);
return Status::OK();
}
- // The analyzer configuration the writers read from the properties, except
that an unset
- // lower_case stays unset, as on the query path; lower_case=true covers
what the writers pass.
- const auto& properties = config.properties;
- InvertedIndexAnalyzerConfig analyzer_config;
- analyzer_config.analyzer_name =
get_analyzer_name_from_properties(properties);
- analyzer_config.parser_type = get_inverted_index_parser_type_from_string(
- get_parser_string_from_properties(properties));
- analyzer_config.parser_mode =
get_parser_mode_string_from_properties(properties);
- analyzer_config.lower_case =
get_parser_lowercase_from_properties(properties);
- analyzer_config.stop_words =
get_parser_stopwords_from_properties(properties);
- analyzer_config.char_filter_map =
get_parser_char_filter_map_from_properties(properties);
+ const InvertedIndexAnalyzerConfig analyzer_config =
analyzer_config_of(config.properties);
try {
*driver = std::make_unique<LuceneDriver>(
- [analyzer_config]() {
- return
InvertedIndexAnalyzer::create_analyzer_provider(&analyzer_config)
- ->get_analyzer();
- },
+ [analyzer_config]() { return build_analyzer(analyzer_config);
},
analyzer_config.char_filter_map);
} catch (const CLuceneError& e) {
return Status::InternalError("building {} failed: {}", config.name,
e.what());
@@ -286,6 +335,31 @@ Status make_golden_driver(const GoldenConfig& config,
std::unique_ptr<GoldenDriv
return Status::OK();
}
+Status make_golden_text_analyzer(const GoldenConfig& config, TextAnalyzerPtr*
analyzer) {
+ DORIS_CHECK(config.kind == GoldenKind::ANALYZE) << config.name;
+ const InvertedIndexAnalyzerConfig analyzer_config =
analyzer_config_of(config.properties);
+ try {
+ *analyzer =
std::make_shared<LuceneBackedTextAnalyzer>(build_analyzer(analyzer_config),
+
analyzer_config.char_filter_map);
+ } catch (const CLuceneError& e) {
+ return Status::InternalError("building {} failed: {}", config.name,
e.what());
+ } catch (const Exception& e) {
+ return Status::InternalError("building {} failed: {}", config.name,
e.what());
+ }
+ return Status::OK();
+}
+
+std::unique_ptr<GoldenDriver> make_text_driver(TextAnalyzerPtr analyzer) {
+ return std::make_unique<TextDriver>(std::move(analyzer));
+}
+
+Status make_golden_text_driver(const GoldenConfig& config,
std::unique_ptr<GoldenDriver>* driver) {
+ TextAnalyzerPtr analyzer;
+ RETURN_IF_ERROR(make_golden_text_analyzer(config, &analyzer));
+ *driver = make_text_driver(std::move(analyzer));
+ return Status::OK();
+}
+
std::vector<std::pair<std::string, std::string>> registered_components() {
auto& factories = AnalysisFactoryMgr::instance();
factories.initialise();
diff --git a/be/test/storage/index/inverted/analysis/golden/golden_driver.h
b/be/test/storage/index/inverted/analysis/golden/golden_driver.h
index e2a445382c7..80f81aba8d0 100644
--- a/be/test/storage/index/inverted/analysis/golden/golden_driver.h
+++ b/be/test/storage/index/inverted/analysis/golden/golden_driver.h
@@ -28,6 +28,7 @@
#include "common/status.h"
#include "storage/index/inverted/analysis/golden/golden_configs.h"
#include "storage/index/inverted/analysis/golden/golden_corpus.h"
+#include "storage/index/inverted/analysis/text_analyzer.h"
namespace lucene::analysis {
class Analyzer;
@@ -100,6 +101,17 @@ std::string format_line(const GoldenLine& line);
// analyzer cannot be built.
Status make_golden_driver(const GoldenConfig& config,
std::unique_ptr<GoldenDriver>* driver);
+// The analyzer of a GoldenKind::ANALYZE `config` behind the TextAnalyzer
interface: the one
+// make_golden_driver() drives, in a LuceneBackedTextAnalyzer.
+Status make_golden_text_analyzer(const GoldenConfig& config, TextAnalyzerPtr*
analyzer);
+
+// Drives `analyzer` as the consumers of the TextAnalyzer interface do: FRESH
makes a stream for
+// every line, as a query does, and REUSED keeps one stream for all lines, as
a writer does.
+std::unique_ptr<GoldenDriver> make_text_driver(TextAnalyzerPtr analyzer);
+
+// make_golden_text_analyzer() in make_text_driver(), with the signature of
make_golden_driver().
+Status make_golden_text_driver(const GoldenConfig& config,
std::unique_ptr<GoldenDriver>* driver);
+
// The kind ("tokenizer", "token_filter" or "char_filter") and name of every
component the
// analysis factory registers.
std::vector<std::pair<std::string, std::string>> registered_components();
diff --git a/be/test/storage/index/inverted/analysis/text_stream_contract.cpp
b/be/test/storage/index/inverted/analysis/text_stream_contract.cpp
new file mode 100644
index 00000000000..3ceba2ba591
--- /dev/null
+++ b/be/test/storage/index/inverted/analysis/text_stream_contract.cpp
@@ -0,0 +1,80 @@
+// 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 "storage/index/inverted/analysis/text_stream_contract.h"
+
+#include <gtest/gtest.h>
+
+namespace doris::segment_v2::inverted_index {
+
+std::ostream& operator<<(std::ostream& out, const CollectedToken& token) {
+ return out << "+" << token.position_increment << "\"" << token.term <<
"\"";
+}
+
+std::vector<CollectedToken> collect_checked(TextTokenStream& stream,
std::string_view text) {
+ constexpr std::string_view kStale = "stale token";
+ std::vector<CollectedToken> tokens;
+ TokenBatch batch;
+ stream.reset(text);
+ for (;;) {
+ batch.push(kStale, 1);
+ if (!stream.next_batch(batch)) {
+ break;
+ }
+ EXPECT_GT(batch.size(), 0) << "next_batch() returned true with no
token";
+ for (const TokenRef& token : batch.tokens()) {
+ EXPECT_GE(token.position_increment, 0);
+ tokens.push_back({.term = std::string(token.term()),
+ .position_increment = token.position_increment});
+ }
+ }
+ EXPECT_EQ(batch.size(), 0) << "next_batch() returned false with tokens in
the batch";
+ batch.push(kStale, 1);
+ EXPECT_FALSE(stream.next_batch(batch)) << "next_batch() returned true
after it returned false";
+ EXPECT_EQ(batch.size(), 0);
+ return tokens;
+}
+
+void expect_text_stream_contract(const TextStreamFactory& make_stream,
+ const std::vector<std::string>& texts) {
+ const std::unique_ptr<TextTokenStream> reused = make_stream();
+ std::vector<std::vector<CollectedToken>> expected;
+ for (const std::string& text : texts) {
+ const std::unique_ptr<TextTokenStream> fresh = make_stream();
+ expected.push_back(collect_checked(*fresh, text));
+ EXPECT_EQ(collect_checked(*reused, text), expected.back())
+ << "a reused stream disagrees with a new one on \"" << text <<
"\"";
+ }
+
+ // NFKC_CF can fold a value to an empty view that does not point to null.
+ const std::string not_null = "x";
+ EXPECT_TRUE(collect_checked(*reused, std::string_view()).empty());
+ EXPECT_TRUE(collect_checked(*reused, std::string_view(not_null.data(),
0)).empty());
+
+ // Each text cut after its first batch, then the next text, which comes
out whole.
+ for (size_t i = 0; i < texts.size(); ++i) {
+ TokenBatch batch;
+ reused->reset(texts[i]);
+ reused->next_batch(batch);
+ const size_t next = (i + 1) % texts.size();
+ EXPECT_EQ(collect_checked(*reused, texts[next]), expected[next])
+ << "\"" << texts[next] << "\" after a reset in the middle of
\"" << texts[i]
+ << "\"";
+ }
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/test/storage/index/inverted/analysis/text_stream_contract.h
b/be/test/storage/index/inverted/analysis/text_stream_contract.h
new file mode 100644
index 00000000000..870bef1fa17
--- /dev/null
+++ b/be/test/storage/index/inverted/analysis/text_stream_contract.h
@@ -0,0 +1,56 @@
+// 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.
+
+#pragma once
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "storage/index/inverted/analysis/text_token_stream.h"
+
+namespace doris::segment_v2::inverted_index {
+
+struct CollectedToken {
+ std::string term;
+ int32_t position_increment = 0;
+
+ bool operator==(const CollectedToken&) const = default;
+};
+
+std::ostream& operator<<(std::ostream& out, const CollectedToken& token);
+
+// Reads `text` with `stream`, checking every next_batch() call against the
TextTokenStream
+// contract: true comes with at least one token and no negative increment,
false with an empty
+// batch, and false again on the call after. The batch holds a stale token
before every call, so a
+// stream that does not clear it fails.
+std::vector<CollectedToken> collect_checked(TextTokenStream& stream,
std::string_view text);
+
+using TextStreamFactory = std::function<std::unique_ptr<TextTokenStream>()>;
+
+// The whole contract, for streams from `make_stream`: one stream reused over
`texts` gives each
+// text the tokens a stream of its own gives it; an empty value gives no
token, whatever its data
+// pointer; and a reset() after the first batch of a value drops the rest of
it, which `texts`
+// exercises when one of them makes more than one batch.
+void expect_text_stream_contract(const TextStreamFactory& make_stream,
+ const std::vector<std::string>& texts);
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/test/storage/index/inverted/analysis/text_token_stream_test.cpp
b/be/test/storage/index/inverted/analysis/text_token_stream_test.cpp
new file mode 100644
index 00000000000..be83112c12c
--- /dev/null
+++ b/be/test/storage/index/inverted/analysis/text_token_stream_test.cpp
@@ -0,0 +1,471 @@
+// 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 <fmt/format.h>
+#include <gtest/gtest.h>
+
+#include <algorithm>
+#include <cctype>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "storage/index/inverted/analysis/per_token_stream.h"
+#include "storage/index/inverted/analysis/scratch.h"
+#include "storage/index/inverted/analysis/text_stream_contract.h"
+#include "storage/index/inverted/analysis/token_batch.h"
+
+namespace doris::segment_v2::inverted_index {
+
+namespace {
+
+TokenRef ref(std::string_view term, int32_t position_increment) {
+ return {.data = term.data(),
+ .size = static_cast<uint32_t>(term.size()),
+ .position_increment = position_increment};
+}
+
+// Words separated by spaces, each built in one reused scratch string: a base
that did not copy the
+// terms would hand out the last word of a batch over and over.
+class WordTokenizer final : public PerTokenTokenizer {
+public:
+ size_t reserved_bytes() const override { return _scratch.capacity(); }
+
+ // next_token() calls after it returned false.
+ int calls_after_end = 0;
+
+protected:
+ void on_reset(std::string_view text) override {
+ _text = text;
+ _position = 0;
+ _ended = false;
+ }
+
+ bool next_token(TokenRef& token) override {
+ if (_ended) {
+ ++calls_after_end;
+ return false;
+ }
+ while (_position < _text.size() && _text[_position] == ' ') {
+ ++_position;
+ }
+ if (_position == _text.size()) {
+ _ended = true;
+ return false;
+ }
+ const size_t end = std::min(_text.find(' ', _position), _text.size());
+ _scratch.assign(_text.substr(_position, end - _position));
+ _position = end;
+ token = ref(_scratch, 1);
+ return true;
+ }
+
+private:
+ std::string_view _text;
+ size_t _position = 0;
+ bool _ended = true;
+ std::string _scratch;
+};
+
+// Each token, then its upper-case twin at the same position: two tokens out
of one.
+class UpperCaseTwinFilter final : public PerTokenFilter {
+public:
+ using PerTokenFilter::PerTokenFilter;
+
+ size_t reserved_bytes() const override {
+ return _twin.capacity() + PerTokenFilter::reserved_bytes();
+ }
+
+protected:
+ void on_reset() override { _twin_pending = false; }
+
+ bool next_token(TokenRef& token) override {
+ if (_twin_pending) {
+ _twin_pending = false;
+ token = ref(_twin, 0);
+ return true;
+ }
+ if (!input_token(token)) {
+ return false;
+ }
+ _twin.assign(token.term());
+ std::ranges::transform(_twin, _twin.begin(),
+ [](char c) { return
static_cast<char>(std::toupper(c)); });
+ _twin_pending = true;
+ return true;
+ }
+
+private:
+ std::string _twin;
+ bool _twin_pending = false;
+};
+
+// Keeps every n-th token and adds the increments of the dropped ones to it,
as a stop filter does.
+// With n above the token count it drops every token.
+class KeepEveryNthFilter final : public PerTokenFilter {
+public:
+ KeepEveryNthFilter(std::unique_ptr<TextTokenStream> in, size_t n)
+ : PerTokenFilter(std::move(in)), _n(n) {}
+
+ size_t reserved_bytes() const override { return
PerTokenFilter::reserved_bytes(); }
+
+ // next_token() calls after it returned false.
+ int calls_after_end = 0;
+
+protected:
+ void on_reset() override {
+ _seen = 0;
+ _ended = false;
+ }
+
+ bool next_token(TokenRef& token) override {
+ if (_ended) {
+ ++calls_after_end;
+ return false;
+ }
+ int32_t dropped = 0;
+ while (input_token(token)) {
+ if (++_seen % _n == 0) {
+ token.position_increment += dropped;
+ return true;
+ }
+ dropped += token.position_increment;
+ }
+ _ended = true;
+ return false;
+ }
+
+private:
+ const size_t _n;
+ size_t _seen = 0;
+ bool _ended = true;
+};
+
+// The characters of each token, one token each. It holds the input token
while it hands out the
+// characters, across next_batch() calls when the token is long, as
word_delimiter holds a token
+// while it hands out its parts.
+class CharacterFilter final : public PerTokenFilter {
+public:
+ using PerTokenFilter::PerTokenFilter;
+
+ size_t reserved_bytes() const override { return
PerTokenFilter::reserved_bytes(); }
+
+protected:
+ void on_reset() override {
+ _current = TokenRef();
+ _next = 0;
+ }
+
+ bool next_token(TokenRef& token) override {
+ while (_next == _current.size) {
+ if (!input_token(_current)) {
+ return false;
+ }
+ _next = 0;
+ }
+ token = ref(std::string_view(_current.data + _next, 1), 1);
+ ++_next;
+ return true;
+ }
+
+private:
+ TokenRef _current;
+ size_t _next = 0;
+};
+
+std::string words(size_t count, std::string_view prefix = "w") {
+ std::string text;
+ for (size_t i = 0; i < count; ++i) {
+ text += fmt::format("{}{} ", prefix, i);
+ }
+ return text;
+}
+
+std::vector<CollectedToken> word_tokens(size_t count, std::string_view prefix
= "w") {
+ std::vector<CollectedToken> tokens;
+ for (size_t i = 0; i < count; ++i) {
+ tokens.push_back({.term = fmt::format("{}{}", prefix, i),
.position_increment = 1});
+ }
+ return tokens;
+}
+
+// The sizes of the batches `stream` hands out for `text`.
+std::vector<size_t> batch_sizes(TextTokenStream& stream, std::string_view
text) {
+ std::vector<size_t> sizes;
+ TokenBatch batch;
+ stream.reset(text);
+ while (stream.next_batch(batch)) {
+ sizes.push_back(batch.size());
+ }
+ return sizes;
+}
+
+std::unique_ptr<TextTokenStream> word_tokenizer() {
+ return std::make_unique<WordTokenizer>();
+}
+
+} // namespace
+
+TEST(TokenBatchTest, PushKeepsTheViewAndPushCopyCopies) {
+ std::string source = "abc";
+ TokenBatch batch;
+ batch.push(source, 1);
+ batch.push_copy(source, 2);
+ char* built = batch.allocate_term(3);
+ std::copy_n("def", 3, built);
+ batch.push({built, 3}, 0);
+ batch.push_copy({}, 1);
+ source[0] = 'X';
+
+ ASSERT_EQ(batch.size(), 4);
+ EXPECT_EQ(batch.tokens()[0].term(), "Xbc");
+ EXPECT_EQ(batch.tokens()[1].term(), "abc");
+ EXPECT_EQ(batch.tokens()[1].position_increment, 2);
+ EXPECT_EQ(batch.tokens()[2].term(), "def");
+ EXPECT_EQ(batch.tokens()[2].position_increment, 0);
+ EXPECT_EQ(batch.tokens()[3].size, 0);
+}
+
+// Term storage grows by adding blocks; a term copied before keeps its address
and its bytes.
+TEST(TokenBatchTest, CopiedTermsStayPutWhileTermStorageGrows) {
+ TokenBatch batch;
+ std::vector<std::string> expected;
+ for (size_t i = 0; i < 10000; ++i) {
+ expected.push_back(fmt::format("term{}-{}", i, std::string(i % 97,
'x')));
+ batch.push_copy(expected.back(), 1);
+ }
+ // Far more than the first 4 KiB block, so the terms are spread over
several blocks.
+ ASSERT_GT(batch._terms.size(), 64 * 4096);
+ ASSERT_EQ(batch.size(), expected.size());
+ for (size_t i = 0; i < expected.size(); ++i) {
+ ASSERT_EQ(batch.tokens()[i].term(), expected[i]) << i;
+ }
+}
+
+TEST(TokenBatchTest, ClearKeepsTheTokenArrayAndTheLargestBlock) {
+ TokenBatch batch;
+ for (size_t i = 0; i < 3000; ++i) {
+ batch.push_copy(std::string(40, static_cast<char>('a' + i % 26)), 1);
+ }
+ const size_t token_capacity = batch._tokens.capacity();
+ const TokenRef* token_array = batch._tokens.data();
+ const size_t term_bytes = batch._terms.size();
+
+ batch.clear();
+ EXPECT_EQ(batch.size(), 0);
+ EXPECT_EQ(batch._tokens.capacity(), token_capacity);
+ // The blocks before the largest are freed, the largest stays.
+ EXPECT_LT(batch._terms.size(), term_bytes);
+ EXPECT_GT(batch._terms.size(), 0);
+
+ // A batch the largest block holds allocates nothing.
+ const size_t kept = batch.capacity_bytes();
+ for (size_t i = 0; i < 100; ++i) {
+ batch.push_copy(std::string(40, 'z'), 1);
+ }
+ EXPECT_EQ(batch.capacity_bytes(), kept);
+ EXPECT_EQ(batch._tokens.data(), token_array);
+}
+
+TEST(TokenBatchTest, ReleaseOversizedKeepsAtMostTheHighWaterMark) {
+ // An ordinary batch keeps its memory.
+ TokenBatch ordinary;
+ for (size_t i = 0; i < TextTokenStream::kBatchTokens; ++i) {
+ ordinary.push_copy("word", 1);
+ }
+ ordinary.clear();
+ const size_t kept = ordinary.capacity_bytes();
+ ordinary.release_oversized();
+ EXPECT_GT(kept, 0);
+ EXPECT_EQ(ordinary.capacity_bytes(), kept);
+
+ // Many tokens, then long terms: each leaves more than the mark behind,
and gives it back.
+ TokenBatch many;
+ for (size_t i = 0; i < 10000; ++i) {
+ many.push("w", 1);
+ }
+ many.clear();
+ ASSERT_GT(many.capacity_bytes(), ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+ many.release_oversized();
+ EXPECT_LE(many.capacity_bytes(), ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+
+ TokenBatch long_terms;
+ long_terms.push_copy(std::string(1 << 20, 'x'), 1);
+ long_terms.clear();
+ ASSERT_GT(long_terms.capacity_bytes(), ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+ long_terms.release_oversized();
+ EXPECT_LE(long_terms.capacity_bytes(), ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+
+ long_terms.push_copy("again", 1);
+ EXPECT_EQ(long_terms.tokens()[0].term(), "again");
+}
+
+TEST(PerTokenStreamTest, TokenizerCopiesEveryTermOutOfItsScratch) {
+ WordTokenizer tokenizer;
+ EXPECT_EQ(collect_checked(tokenizer, " alpha beta gamma "),
+ (std::vector<CollectedToken> {{.term = "alpha",
.position_increment = 1},
+ {.term = "beta",
.position_increment = 1},
+ {.term = "gamma",
.position_increment = 1}}));
+ EXPECT_EQ(collect_checked(tokenizer, words(1000)), word_tokens(1000));
+}
+
+TEST(PerTokenStreamTest, BatchesHoldAtMostKBatchTokens) {
+ WordTokenizer tokenizer;
+ EXPECT_EQ(batch_sizes(tokenizer, words(1000)), (std::vector<size_t> {256,
256, 256, 232}));
+ // A value of exactly one batch: the next call finds it exhausted and
returns false.
+ EXPECT_EQ(batch_sizes(tokenizer, words(256)), (std::vector<size_t> {256}));
+ EXPECT_TRUE(batch_sizes(tokenizer, " ").empty());
+}
+
+TEST(PerTokenStreamTest, NextTokenIsNotCalledAgainAfterItReturnedFalse) {
+ WordTokenizer tokenizer;
+ TokenBatch batch;
+ tokenizer.reset("a b");
+ EXPECT_TRUE(tokenizer.next_batch(batch));
+ EXPECT_FALSE(tokenizer.next_batch(batch));
+ EXPECT_FALSE(tokenizer.next_batch(batch));
+ EXPECT_EQ(tokenizer.calls_after_end, 0);
+
+ auto filter = std::make_unique<KeepEveryNthFilter>(word_tokenizer(), 2);
+ const std::string ten = words(10);
+ filter->reset(ten);
+ while (filter->next_batch(batch)) {
+ }
+ EXPECT_FALSE(filter->next_batch(batch));
+ EXPECT_EQ(filter->calls_after_end, 0);
+}
+
+TEST(PerTokenStreamTest, NothingComesOutBeforeAValueIsBound) {
+ WordTokenizer tokenizer;
+ TokenBatch batch;
+ EXPECT_FALSE(tokenizer.next_batch(batch));
+ UpperCaseTwinFilter filter(word_tokenizer());
+ EXPECT_FALSE(filter.next_batch(batch));
+ EXPECT_EQ(batch.size(), 0);
+}
+
+TEST(PerTokenStreamTest, FilterMakesSeveralTokensOutOfOne) {
+ UpperCaseTwinFilter filter(word_tokenizer());
+ const std::vector<CollectedToken> tokens = collect_checked(filter,
words(300));
+ ASSERT_EQ(tokens.size(), 600);
+ for (size_t i = 0; i < 300; ++i) {
+ EXPECT_EQ(tokens[2 * i],
+ (CollectedToken {.term = fmt::format("w{}", i),
.position_increment = 1}));
+ EXPECT_EQ(tokens[2 * i + 1],
+ (CollectedToken {.term = fmt::format("W{}", i),
.position_increment = 0}));
+ }
+ EXPECT_EQ(batch_sizes(filter, words(300)), (std::vector<size_t> {256, 256,
88}));
+}
+
+// A filter that drops every token of a batch keeps pulling its input: no call
returns true with an
+// empty batch, and a value whose tokens are all dropped makes the first call
return false.
+TEST(PerTokenStreamTest, FilterThatDropsTokensReturnsNoEmptyBatch) {
+ KeepEveryNthFilter drop_all(word_tokenizer(), 1000);
+ TokenBatch batch;
+ const std::string text = words(999);
+ drop_all.reset(text);
+ EXPECT_FALSE(drop_all.next_batch(batch));
+ EXPECT_EQ(batch.size(), 0);
+
+ // Three of 1000 tokens survive, from input batches far apart.
+ KeepEveryNthFilter keep_few(word_tokenizer(), 300);
+ EXPECT_EQ(collect_checked(keep_few, words(1000)),
+ (std::vector<CollectedToken> {{.term = "w299",
.position_increment = 300},
+ {.term = "w599",
.position_increment = 300},
+ {.term = "w899",
.position_increment = 300}}));
+}
+
+// The filter hands out the characters of a 300-character token across two
batches while it holds
+// the token, and the input batch the token lives in stays as it is meanwhile.
+TEST(PerTokenStreamTest, FilterHoldsAnInputTokenAcrossBatches) {
+ const std::string first(300, 'a');
+ const std::string second = "bcd";
+ CharacterFilter filter(word_tokenizer());
+ EXPECT_EQ(batch_sizes(filter, first + " " + second), (std::vector<size_t>
{256, 47}));
+ const std::vector<CollectedToken> tokens = collect_checked(filter, first +
" " + second);
+ ASSERT_EQ(tokens.size(), 303);
+ for (size_t i = 0; i < 300; ++i) {
+ EXPECT_EQ(tokens[i].term, "a") << i;
+ }
+ EXPECT_EQ(tokens[300].term, "b");
+ EXPECT_EQ(tokens[302].term, "d");
+}
+
+TEST(PerTokenStreamTest, ResetDropsTheRestOfAValue) {
+ TokenBatch batch;
+ WordTokenizer tokenizer;
+ // A value must stay valid while the stream reads it, so it is not a
temporary.
+ const std::string long_text = words(1000);
+ tokenizer.reset(long_text);
+ ASSERT_TRUE(tokenizer.next_batch(batch));
+ EXPECT_EQ(collect_checked(tokenizer, "x y"),
+ (std::vector<CollectedToken> {{.term = "x", .position_increment
= 1},
+ {.term = "y", .position_increment
= 1}}));
+
+ UpperCaseTwinFilter twins(word_tokenizer());
+ twins.reset(long_text);
+ ASSERT_TRUE(twins.next_batch(batch));
+ EXPECT_EQ(collect_checked(twins, "x"),
+ (std::vector<CollectedToken> {{.term = "x", .position_increment
= 1},
+ {.term = "X", .position_increment
= 0}}));
+
+ // Cut in the middle of a held token.
+ CharacterFilter characters(word_tokenizer());
+ const std::string long_word(300, 'a');
+ characters.reset(long_word);
+ ASSERT_TRUE(characters.next_batch(batch));
+ EXPECT_EQ(collect_checked(characters, "bc"),
+ (std::vector<CollectedToken> {{.term = "b", .position_increment
= 1},
+ {.term = "c", .position_increment
= 1}}));
+}
+
+TEST(PerTokenStreamTest,
ReservedBytesAddUpTheChainAndResetReleasesTheInputBatch) {
+ UpperCaseTwinFilter filter(word_tokenizer());
+ // One word of 1 MiB: the input batch copies it, so it holds more than the
mark.
+ const std::string huge(1 << 20, 'h');
+ TokenBatch batch;
+ filter.reset(huge);
+ ASSERT_TRUE(filter.next_batch(batch));
+ const auto& tokenizer = static_cast<const WordTokenizer&>(*filter._in);
+ EXPECT_GT(filter._input.capacity_bytes(),
ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+ EXPECT_EQ(filter.reserved_bytes(), filter._twin.capacity() +
filter._input.capacity_bytes() +
+ tokenizer.reserved_bytes());
+
+ filter.reset("short");
+ EXPECT_LE(filter._input.capacity_bytes(),
ANALYZER_SCRATCH_HIGH_WATER_BYTES);
+ EXPECT_EQ(filter.reserved_bytes(), filter._twin.capacity() +
filter._input.capacity_bytes() +
+ tokenizer.reserved_bytes());
+}
+
+TEST(PerTokenStreamTest, TestStreamsKeepTheContract) {
+ const std::vector<std::string> texts = {
+ "one", " a b c ", words(1000), words(256), std::string(300, 'a'),
"x"};
+ expect_text_stream_contract(word_tokenizer, texts);
+ expect_text_stream_contract(
+ [] { return
std::make_unique<UpperCaseTwinFilter>(word_tokenizer()); }, texts);
+ expect_text_stream_contract(
+ [] { return std::make_unique<KeepEveryNthFilter>(word_tokenizer(),
3); }, texts);
+ expect_text_stream_contract([] { return
std::make_unique<CharacterFilter>(word_tokenizer()); },
+ texts);
+ // Every token dropped, for every text.
+ expect_text_stream_contract(
+ [] { return std::make_unique<KeepEveryNthFilter>(word_tokenizer(),
100000); }, texts);
+}
+
+} // namespace doris::segment_v2::inverted_index
diff --git a/be/test/storage/index/inverted/clucene/lucene_adapter_test.cpp
b/be/test/storage/index/inverted/clucene/lucene_adapter_test.cpp
new file mode 100644
index 00000000000..9d19460ef3c
--- /dev/null
+++ b/be/test/storage/index/inverted/clucene/lucene_adapter_test.cpp
@@ -0,0 +1,455 @@
+// 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.
+
+// The two adapters between the CLucene analysis interfaces and the
TextAnalyzer interface:
+// LuceneBackedTextAnalyzer, a CLucene analyzer for the new consumers, and
LuceneAnalyzerBridge, a
+// TextAnalyzer for the V1 and V2 writers.
+
+#include <CLucene.h>
+#include <fmt/format.h>
+#include <gen_cpp/olap_file.pb.h>
+#include <gtest/gtest.h>
+
+#include <climits>
+#include <functional>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/exception.h"
+#include "common/status.h"
+#include "io/fs/local_file_system.h"
+#include "storage/index/index_file_writer.h"
+#include "storage/index/index_writer.h"
+#include "storage/index/inverted/analysis/analyzer/analyzer.h"
+#include "storage/index/inverted/analysis/inverted_index_parser.h"
+#include "storage/index/inverted/analysis/text_stream_contract.h"
+#include "storage/index/inverted/clucene/inverted_index_writer.h"
+#include "storage/index/inverted/clucene/lucene_analyzer_bridge.h"
+#include "storage/index/inverted/clucene/lucene_backed_text_analyzer.h"
+#include "storage/index/inverted/common/inverted_index_desc.h"
+#include "storage/tablet/tablet_schema.h"
+#include "util/slice.h"
+
+namespace doris::segment_v2::inverted_index {
+
+namespace {
+
+constexpr const char* kTestDir = "./ut_dir/lucene_adapter_test";
+
+using Tokens = std::vector<CollectedToken>;
+
+std::string words(size_t count) {
+ std::string text;
+ for (size_t i = 0; i < count; ++i) {
+ text += fmt::format("w{} ", i);
+ }
+ return text;
+}
+
+std::string read_all(lucene::util::Reader* reader) {
+ const char* start = nullptr;
+ const int32_t length = reader->read(reinterpret_cast<const
void**>(&start), 0,
+ static_cast<int32_t>(reader->size()));
+ return length > 0 ? std::string(start, length) : std::string();
+}
+
+// A word of a script value: words are separated by spaces, "_" is an empty
term, a leading '^'
+// puts the word at the position of the one before, and "!" is a failure where
it stands.
+struct ScriptWord {
+ std::string_view term;
+ int32_t position_increment = 1;
+ bool fails = false;
+};
+
+// The next word of `text` from `*position`, or false at the end.
+bool next_script_word(std::string_view text, size_t* position, ScriptWord*
word) {
+ while (*position < text.size() && text[*position] == ' ') {
+ ++*position;
+ }
+ if (*position == text.size()) {
+ return false;
+ }
+ const size_t end = std::min(text.find(' ', *position), text.size());
+ std::string_view term = text.substr(*position, end - *position);
+ *position = end;
+ *word = ScriptWord();
+ if (term == "!") {
+ word->fails = true;
+ return true;
+ }
+ if (term.starts_with('^')) {
+ word->position_increment = 0;
+ term.remove_prefix(1);
+ }
+ word->term = term == "_" ? term.substr(0, 0) : term;
+ return true;
+}
+
+// A CLucene stream over a script value; "!" throws a CLuceneError.
+class ScriptedLuceneStream final : public lucene::analysis::TokenStream {
+public:
+ explicit ScriptedLuceneStream(std::string text) : _text(std::move(text)) {}
+
+ lucene::analysis::Token* next(lucene::analysis::Token* token) override {
+ ScriptWord word;
+ if (!next_script_word(_text, &_position, &word)) {
+ return nullptr;
+ }
+ if (word.fails) {
+ _CLTHROWA(CL_ERR_IO, "scripted CLucene failure");
+ }
+ token->setTextNoCopy(word.term.data(),
static_cast<int32_t>(word.term.size()));
+ token->setPositionIncrement(word.position_increment);
+ return token;
+ }
+
+ void close() override {}
+
+private:
+ const std::string _text;
+ size_t _position = 0;
+};
+
+// A CLucene analyzer of script values; the value "!!" makes tokenStream()
itself throw.
+class ScriptedLuceneAnalyzer final : public lucene::analysis::Analyzer {
+public:
+ bool isSDocOpt() override { return true; }
+
+ lucene::analysis::TokenStream* tokenStream(const TCHAR*, const ReaderPtr&
reader) override {
+ std::string text = read_all(reader.get());
+ if (text == "!!") {
+ _CLTHROWA(CL_ERR_IO, "scripted CLucene failure in tokenStream");
+ }
+ return new ScriptedLuceneStream(std::move(text));
+ }
+
+ lucene::analysis::TokenStream* tokenStream(const TCHAR*,
lucene::util::Reader*) override {
+ throw Exception(ErrorCode::INVERTED_INDEX_NOT_SUPPORTED, "not used");
+ }
+
+ lucene::analysis::TokenStream* reusableTokenStream(const TCHAR*,
+ lucene::util::Reader*)
override {
+ throw Exception(ErrorCode::INVERTED_INDEX_NOT_SUPPORTED, "not used");
+ }
+
+ lucene::analysis::TokenStream* reusableTokenStream(const TCHAR*, const
ReaderPtr&) override {
+ throw Exception(ErrorCode::INVERTED_INDEX_NOT_SUPPORTED, "not used");
+ }
+};
+
+// A native stream over a script value, handing out views of it; "!" throws a
doris::Exception
+// from next_batch(), and a value starting with "!reset" throws from reset().
+class ScriptedTextStream final : public TextTokenizer {
+public:
+ void reset(std::string_view text) override {
+ if (text.starts_with("!reset")) {
+ throw Exception(ErrorCode::INVERTED_INDEX_ANALYZER_ERROR,
"scripted failure in reset");
+ }
+ _text = text;
+ _position = 0;
+ }
+
+ bool next_batch(TokenBatch& batch) override {
+ batch.clear();
+ ScriptWord word;
+ while (batch.size() < kBatchTokens && next_script_word(_text,
&_position, &word)) {
+ if (word.fails) {
+ throw Exception(ErrorCode::INVERTED_INDEX_ANALYZER_ERROR,
"scripted failure");
+ }
+ batch.push(word.term, word.position_increment);
+ }
+ return batch.size() != 0;
+ }
+
+ size_t reserved_bytes() const override { return 0; }
+
+private:
+ std::string_view _text;
+ size_t _position = 0;
+};
+
+class ScriptedTextAnalyzer final : public TextAnalyzer {
+public:
+ std::unique_ptr<TextTokenStream> create_stream() const override {
+ return std::make_unique<ScriptedTextStream>();
+ }
+};
+
+const Tokens kScriptTokens = {{.term = "a", .position_increment = 1},
+ {.term = "b", .position_increment = 0},
+ {.term = "", .position_increment = 1},
+ {.term = "c", .position_increment = 1}};
+
+ReaderPtr reader_of(const std::string& text) {
+ auto reader = std::make_shared<lucene::util::SStringReader<char>>();
+ reader->init(text.data(), static_cast<int32_t>(text.size()), false);
+ return reader;
+}
+
+// The tokens of a bridged stream, read as SDocumentsWriter::invertField()
reads a field: reset(),
+// next() until it returns null with one token object, close().
+Tokens lucene_tokens(lucene::analysis::TokenStream* stream) {
+ Tokens tokens;
+ lucene::analysis::Token token;
+ stream->reset();
+ while (stream->next(&token) != nullptr) {
+ tokens.push_back({.term = std::string(token.termBuffer<char>(),
token.termLength<char>()),
+ .position_increment = token.getPositionIncrement()});
+ }
+ stream->close();
+ return tokens;
+}
+
+// The code of the doris::Exception `body` throws, or OK.
+int thrown_code(const std::function<void()>& body) {
+ try {
+ body();
+ } catch (const Exception& e) {
+ return e.code();
+ }
+ return ErrorCode::OK;
+}
+
+} // namespace
+
+TEST(LuceneBackedTextAnalyzerTest, HandsOutTheCLuceneTokens) {
+ const LuceneBackedTextAnalyzer
analyzer(std::make_shared<ScriptedLuceneAnalyzer>(), {});
+ const auto stream = analyzer.create_stream();
+ EXPECT_EQ(collect_checked(*stream, "a ^b _ c"), kScriptTokens);
+ EXPECT_EQ(stream->reserved_bytes(), 0);
+
+ TokenBatch batch;
+ std::vector<size_t> sizes;
+ const std::string long_text = words(1000);
+ stream->reset(long_text);
+ while (stream->next_batch(batch)) {
+ sizes.push_back(batch.size());
+ }
+ EXPECT_EQ(sizes, (std::vector<size_t> {256, 256, 256, 232}));
+}
+
+TEST(LuceneBackedTextAnalyzerTest, ReadsTheValueThroughTheIndexCharFilter) {
+ const LuceneBackedTextAnalyzer analyzer(
+ std::make_shared<ScriptedLuceneAnalyzer>(),
+ {{INVERTED_INDEX_PARSER_CHAR_FILTER_TYPE,
INVERTED_INDEX_CHAR_FILTER_CHAR_REPLACE},
+ {INVERTED_INDEX_PARSER_CHAR_FILTER_PATTERN, "-"},
+ {INVERTED_INDEX_PARSER_CHAR_FILTER_REPLACEMENT, " "}});
+ EXPECT_EQ(collect_checked(*analyzer.create_stream(), "a-b"),
+ (Tokens {{.term = "a", .position_increment = 1},
+ {.term = "b", .position_increment = 1}}));
+}
+
+TEST(LuceneBackedTextAnalyzerTest, CLuceneErrorsComeOutAsDorisExceptions) {
+ const LuceneBackedTextAnalyzer
analyzer(std::make_shared<ScriptedLuceneAnalyzer>(), {});
+ const auto stream = analyzer.create_stream();
+ TokenBatch batch;
+ EXPECT_EQ(thrown_code([&] {
+ stream->reset("a b !");
+ while (stream->next_batch(batch)) {
+ }
+ }),
+ ErrorCode::INVERTED_INDEX_ANALYZER_ERROR);
+ EXPECT_EQ(thrown_code([&] { stream->reset("!!"); }),
ErrorCode::INVERTED_INDEX_ANALYZER_ERROR);
+ // The next value starts over.
+ EXPECT_EQ(collect_checked(*stream, "c"), (Tokens {{.term = "c",
.position_increment = 1}}));
+}
+
+TEST(LuceneBackedTextAnalyzerTest, KeepsTheTextStreamContract) {
+ const LuceneBackedTextAnalyzer
scripted(std::make_shared<ScriptedLuceneAnalyzer>(), {});
+ expect_text_stream_contract([&] { return scripted.create_stream(); },
+ {"a ^b _ c", words(1000), words(256), "x"});
+ const LuceneBackedTextAnalyzer standard(
+
InvertedIndexAnalyzer::create_builtin_analyzer(InvertedIndexParserType::PARSER_STANDARD,
+ "", "", ""),
+ {});
+ expect_text_stream_contract([&] { return standard.create_stream(); },
+ {"The Quick brown fox", words(1000), "中文 mixed
文本", "x"});
+}
+
+TEST(LuceneAnalyzerBridgeTest, HandsOutTheTokensOfTheTextAnalyzer) {
+ LuceneAnalyzerBridge bridge(std::make_shared<ScriptedTextAnalyzer>());
+ const std::string first = "a ^b _ c";
+ lucene::analysis::TokenStream* stream = bridge.reusableTokenStream(L"f",
reader_of(first));
+ EXPECT_EQ(lucene_tokens(stream), kScriptTokens);
+
+ // A scalar column rebinds the same stream for every value.
+ const std::string second = words(1000);
+ EXPECT_EQ(bridge.reusableTokenStream(L"f", reader_of(second)), stream);
+ const Tokens tokens = lucene_tokens(stream);
+ ASSERT_EQ(tokens.size(), 1000);
+ EXPECT_EQ(tokens[999].term, "w999");
+
+ const std::string empty;
+ EXPECT_TRUE(lucene_tokens(bridge.reusableTokenStream(L"f",
reader_of(empty))).empty());
+}
+
+// invertField() calls reset() before it reads a field. Before that, next()
hands out nothing, as
+// DorisTokenizer does before reset() promotes its reader; a second reset()
starts the value over.
+TEST(LuceneAnalyzerBridgeTest, ResetStartsTheValue) {
+ LuceneAnalyzerBridge bridge(std::make_shared<ScriptedTextAnalyzer>());
+ const std::string value = "a b";
+ lucene::analysis::TokenStream* stream = bridge.reusableTokenStream(L"f",
reader_of(value));
+ lucene::analysis::Token token;
+ EXPECT_EQ(stream->next(&token), nullptr);
+ EXPECT_EQ(lucene_tokens(stream).size(), 2);
+ EXPECT_EQ(lucene_tokens(stream).size(), 2);
+}
+
+TEST(LuceneAnalyzerBridgeTest, TokenStreamMakesAStreamTheCallerOwns) {
+ LuceneAnalyzerBridge bridge(std::make_shared<ScriptedTextAnalyzer>());
+ const std::string first = "a b";
+ const std::string second = "c";
+ const std::unique_ptr<lucene::analysis::TokenStream> one(
+ bridge.tokenStream(L"f", reader_of(first)));
+ const std::unique_ptr<lucene::analysis::TokenStream> two(
+ bridge.tokenStream(L"f", reader_of(second)));
+ EXPECT_NE(one.get(), two.get());
+ EXPECT_EQ(lucene_tokens(two.get()), (Tokens {{.term = "c",
.position_increment = 1}}));
+ EXPECT_EQ(lucene_tokens(one.get()), (Tokens {{.term = "a",
.position_increment = 1},
+ {.term = "b",
.position_increment = 1}}));
+}
+
+TEST(LuceneAnalyzerBridgeTest, ReaderPointerOverloadsThrow) {
+ LuceneAnalyzerBridge bridge(std::make_shared<ScriptedTextAnalyzer>());
+ const std::string value = "a";
+ const ReaderPtr reader = reader_of(value);
+ EXPECT_EQ(thrown_code([&] { bridge.tokenStream(L"f", reader.get()); }),
+ ErrorCode::INVERTED_INDEX_NOT_SUPPORTED);
+ EXPECT_EQ(thrown_code([&] { bridge.reusableTokenStream(L"f",
reader.get()); }),
+ ErrorCode::INVERTED_INDEX_NOT_SUPPORTED);
+}
+
+// A doris::Exception from a bridged stream passes through the CLucene frames
of
+// IndexWriter::addDocument() to the V2 writer, which reports it, for a string
column and for an
+// array column, thrown from reset() and from next_batch(). LeakSanitizer
checks that the CLucene
+// state the exception skipped over is freed.
+class LuceneBridgeWriterTest : public testing::Test {
+protected:
+ void SetUp() override {
+
ASSERT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+
ASSERT_TRUE(io::global_local_filesystem()->create_directory(kTestDir).ok());
+ TabletIndexPB pb;
+ pb.set_index_type(IndexType::INVERTED);
+ pb.set_index_id(91);
+ pb.set_index_name("bridge_writer_idx");
+ pb.add_col_unique_id(0);
+ pb.mutable_properties()->insert({"parser", "english"});
+ pb.mutable_properties()->insert({"support_phrase", "true"});
+ _meta.init_from_pb(pb);
+ }
+
+ void TearDown() override {
+ _writer.reset();
+ _index_file_writer.reset();
+ _kept_analyzers.clear();
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ }
+
+ // A V2 writer for `column` whose analyzer is the bridge over
ScriptedTextAnalyzer, in place of
+ // the writer before.
+ void make_writer(const TabletColumn& column, const std::string& name) {
+ _writer.reset();
+ _index_file_writer.reset();
+ const std::string prefix = std::string(kTestDir) + "/" + name;
+ const auto fs = io::global_local_filesystem();
+ io::FileWriterPtr file_writer;
+
ASSERT_TRUE(fs->create_file(InvertedIndexDescriptor::get_index_file_path_v2(prefix),
+ &file_writer)
+ .ok());
+ _index_file_writer = std::make_unique<IndexFileWriter>(
+ fs, prefix, "bridge_rowset", /*seg_id=*/0,
InvertedIndexStorageFormatPB::V2,
+ std::move(file_writer));
+ const Status created =
+ IndexColumnWriter::create(&column, &_writer,
_index_file_writer.get(), &_meta);
+ ASSERT_TRUE(created.ok()) << created.to_string();
+ auto* legacy =
static_cast<InvertedIndexColumnWriter<FieldType::OLAP_FIELD_TYPE_STRING>*>(
+ _writer.get());
+ // The CLucene IndexWriter keeps a raw pointer to the analyzer it was
made with.
+ _kept_analyzers.push_back(legacy->_analyzer);
+ legacy->set_analysis_for_test(
+ std::make_shared<lucene::util::SStringReader<char>>(),
+
std::make_shared<LuceneAnalyzerBridge>(std::make_shared<ScriptedTextAnalyzer>()));
+ }
+
+ static TabletColumn text_column() {
+ TabletColumn column;
+ column.set_unique_id(0);
+ column.set_name("text");
+ column.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
+ column.set_length(INT_MAX);
+ column.set_is_nullable(true);
+ return column;
+ }
+
+ static TabletColumn array_column() {
+ TabletColumn element = text_column();
+ TabletColumn column;
+ column.set_unique_id(0);
+ column.set_name("texts");
+ column.set_type(FieldType::OLAP_FIELD_TYPE_ARRAY);
+ column.set_is_nullable(true);
+ column.add_sub_column(element);
+ return column;
+ }
+
+ TabletIndex _meta;
+ std::vector<std::shared_ptr<lucene::analysis::Analyzer>> _kept_analyzers;
+ std::unique_ptr<IndexFileWriter> _index_file_writer;
+ std::unique_ptr<IndexColumnWriter> _writer;
+};
+
+TEST_F(LuceneBridgeWriterTest, AnExceptionFromTheStreamFailsAddValues) {
+ for (const std::string bad : {"x ! y", "!reset z"}) {
+ make_writer(text_column(), "scalar_" + std::to_string(bad.size()));
+ const std::vector<std::string> rows = {"fine words", bad};
+ const std::vector<Slice> slices = {Slice(rows[0]), Slice(rows[1])};
+ const Status status =
+ _writer->add_values(reinterpret_cast<const
uint8_t*>(slices.data()), slices.size());
+ EXPECT_EQ(status.code(), ErrorCode::INVERTED_INDEX_ANALYZER_ERROR)
+ << bad << ": " << status.to_string();
+ // Caught around IndexWriter::addDocument(), not where the field is
made.
+ EXPECT_NE(status.to_string().find("while adding document"),
std::string::npos)
+ << status.to_string();
+ EXPECT_NE(status.to_string().find("scripted failure"),
std::string::npos)
+ << status.to_string();
+ }
+}
+
+TEST_F(LuceneBridgeWriterTest, AnExceptionFromTheStreamFailsAddArrayValues) {
+ for (const std::string bad : {"x ! y", "!reset z"}) {
+ make_writer(array_column(), "array_" + std::to_string(bad.size()));
+ const std::vector<std::string> elements = {"fine words", bad};
+ const std::vector<Slice> slices = {Slice(elements[0]),
Slice(elements[1])};
+ const std::vector<uint8_t> null_map = {0, 0};
+ const std::vector<uint64_t> offsets = {0, 2};
+ const Status status = _writer->add_array_values(
+ sizeof(Slice), reinterpret_cast<const
uint8_t*>(slices.data()), null_map.data(),
+ offsets.data(), 1);
+ EXPECT_EQ(status.code(), ErrorCode::INVERTED_INDEX_ANALYZER_ERROR)
+ << bad << ": " << status.to_string();
+ // Caught around IndexWriter::addDocument(), not where the field is
made.
+ EXPECT_NE(status.to_string().find("while adding document"),
std::string::npos)
+ << status.to_string();
+ EXPECT_NE(status.to_string().find("scripted failure"),
std::string::npos)
+ << status.to_string();
+ }
+}
+
+} // namespace doris::segment_v2::inverted_index
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
