eldenmoon commented on code in PR #69761:
URL: https://github.com/apache/doris/pull/69761#discussion_r4438597383
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/IndexDefinition.java:
##########
@@ -562,6 +572,66 @@ public Map<String, String> getProperties() {
return properties;
}
+ public boolean isVariantValuesIndex() {
+ return indexType == IndexType.INVERTED &&
InvertedIndexProperties.isVariantValuesIndex(properties);
+ }
+
+ private void validateVariantValuesIndex(DataType columnType,
+ TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat) {
+ String scope =
properties.get(InvertedIndexUtil.VARIANT_INDEX_SCOPE_KEY);
+ if (scope == null) {
+ return;
+ }
+ if (!isVariantValuesIndex()) {
+ throw new AnalysisException("variant_index_scope must be values,
found: " + scope);
+ }
+ if (!columnType.isVariantType()) {
+ throw new AnalysisException(
+ "variant_index_scope=values can only be used on VARIANT
columns");
+ }
+ // Declared Decimal256 can be stored, but full-root assembly cannot
decode it.
+ // Reject this combination before accepting a table whose legal
inserts would fail.
+ VariantType variantType = (VariantType) columnType;
+ if (variantType.getEnableNestedGroup()) {
+ throw new AnalysisException("VARIANT values index does not support
NestedGroup");
+ }
+ if (variantType.getPredefinedFields().stream()
+ .anyMatch(field -> containsDecimal256(field.getDataType()))) {
+ throw new AnalysisException("VARIANT values index does not support
declared Decimal256 paths");
+ }
+ if (invertedIndexFileStorageFormat !=
TInvertedIndexFileStorageFormat.SNII) {
+ throw new AnalysisException(
+ "VARIANT values index requires
inverted_index_storage_format=SNII");
+ }
+ if
("true".equals(properties.get(InvertedIndexUtil.INVERTED_INDEX_SUPPORT_PHRASE_KEY)))
{
+ throw new AnalysisException("VARIANT values index does not support
support_phrase=true");
+ }
+ String analyzer =
properties.getOrDefault(InvertedIndexUtil.INVERTED_INDEX_ANALYZER_NAME_KEY,
"").trim();
+ if
(InvertedIndexProperties.INVERTED_INDEX_PARSER_NONE.equalsIgnoreCase(
+ InvertedIndexProperties.getInvertedIndexParser(properties))
+ && (analyzer.isEmpty()
+ ||
InvertedIndexProperties.INVERTED_INDEX_PARSER_NONE.equalsIgnoreCase(analyzer)))
{
+ throw new AnalysisException("VARIANT values index requires a
parser or analyzer with tokenization");
+ }
+ if (!analyzer.isEmpty()
+ &&
!IndexPolicy.BUILTIN_ANALYZERS.contains(InvertedIndexUtil.resolveAnalyzerName(analyzer)))
{
+ throw new AnalysisException("VARIANT values index supports only
built-in parsers and analyzers");
+ }
+ if
(properties.containsKey(InvertedIndexUtil.INVERTED_INDEX_PARSER_FIELD_PATTERN_KEY))
{
+ throw new AnalysisException("VARIANT values index does not support
field_pattern");
+ }
+ }
+
+ private static boolean containsDecimal256(DataType type) {
+ if (type instanceof DecimalV3Type) {
+ return ((DecimalV3Type) type).getPrecision() > 38;
+ }
+ if (type instanceof ArrayType) {
Review Comment:
Not reachable. VARIANT predefined fields accept only scalar types and one
level of ARRAY over them (`VariantType.supportSubType`,
`fe/fe-type/.../VariantType.java:215`, enforced at DDL by
`DataType.validateNestedType`). `VARIANT<'obj':STRUCT<...>>`,
`VARIANT<'m':MAP<...>>`, `ARRAY<MAP<...>>` and a nested `VARIANT<...>` are all
rejected with `VARIANT unsupported sub-type` before the index is checked, so
`containsDecimal256()` already covers every legal declared type (direct and
array).
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/InvertedIndexUtil.java:
##########
@@ -504,9 +519,18 @@ public static boolean
canHaveMultipleInvertedIndexes(DataType colType, List<Inde
return false;
}
+ if
(indexDefs.stream().filter(IndexDefinition::isVariantRootIndex).count() > 1) {
+ return false;
+ }
+
Set<String> analyzerKeys = new HashSet<>();
Set<String> analyzerSelectors = new HashSet<>();
for (IndexDefinition indexDef : indexDefs) {
+ // Root and path indexes bind disjoint predicates. Only path
indexes share
+ // analyzer identities/selectors; the single-root constraint is
checked above.
+ if (indexDef.isVariantRootIndex()) {
Review Comment:
Resolved at this head. On a table with an English values index, `ALTER TABLE
t ADD INDEX path_idx(v) USING INVERTED PROPERTIES("parser"="english")`
succeeds, and `v['s'] MATCH_ANY` uses it for new rowsets, the same as on a
table without a values index (older rowsets need BUILD INDEX in both cases).
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CheckMatchExpression.java:
##########
@@ -56,9 +57,22 @@ private Plan checkChildren(LogicalFilter<? extends Plan>
filter) {
+ " right operand is Literal. But meet expression %s",
matchExpression));
}
if (slotReference.getDataType().isVariantType() &&
!slotReference.hasSubColPath()) {
- throw new AnalysisException(String.format("VARIANT root column
does not support MATCH predicates. "
- + "Please query a subcolumn instead, for
example %s['field'] MATCH 'xxx'",
- slotReference.getName()));
+ if (matchExpression.getAnalyzer().isPresent()
+ &&
!"english".equals(matchExpression.getAnalyzer().get())) {
Review Comment:
Resolved at this head. On an English values-indexed table, `v MATCH_ANY
'hello' USING ANALYZER ENGLISH` (and `English`) is accepted in both WHERE and
the projection and returns the same rows as the lower-case spelling.
##########
be/src/storage/segment/variant/v2/variant_shredder.cpp:
##########
@@ -613,6 +617,88 @@ struct VariantShredder::Impl {
return publish_materialized(selection, candidate_builders, result);
}
+ // Typed paths are indexed only after the same conversion used by storage.
Merge compact
+ // rowids instead of scanning every path for every row or reconstructing
complete objects.
+ //
NOLINTNEXTLINE(readability-function-size,readability-function-cognitive-complexity,readability-make-member-function-const):
keep encoded leaf ownership and synchronous writer consumption in one row
scope; the root index writers are mutated.
+ Status write_converted_root_indexes(const VariantShreddedColumns& result,
+ const
DorisVector<VariantPathBuilder*>& builders) {
+ using namespace variant_v2::variant_assembler_detail;
+ using Position = std::pair<size_t, size_t>;
+ std::priority_queue<Position, DorisVector<Position>, std::greater<>>
pending;
+ DorisVector<PreparedMaterializedColumn> values;
+ DorisVector<size_t> positions(builders.size(), 0);
+ values.reserve(builders.size());
+ for (size_t index = 0; index < builders.size() &&
+ options.physical_layout !=
VariantShredderPhysicalLayout::DOC;
+ ++index) {
+ const auto* builder = builders[index];
+ if (builder->rowids().empty()) {
+ values.emplace_back();
+ continue;
+ }
+ values.push_back(prepare_materialized_column(builder->type(),
builder->column().get(),
+
builder->rowids().size()));
+ pending.emplace(builder->rowids().front(), index);
+ }
+ const auto& roots = assert_cast<const
ColumnString&>(*result.root_jsonb);
+ DorisVector<size_t> row_paths;
+ row_paths.reserve(builders.size() + 1);
+ for (size_t row = 0; row < rows; ++row) {
+ row_paths.clear();
+ VariantBatchBuilder encoded;
+ if (!index_nulls[row] && !roots.get_data_at(row).empty()) {
+ auto output = encoded.begin_row();
+ jsonb_to_variant(roots.get_data_at(row), output, 0);
+ output.finish();
+ row_paths.push_back(builders.size());
+ }
+ if (!index_nulls[row] &&
+ options.physical_layout == VariantShredderPhysicalLayout::DOC)
{
+ // DOC root reads the finalized binary buckets, not converted
typed sidecars.
+ for (const auto& bucket : result.binary_buckets) {
+ const auto& map = assert_cast<const
ColumnMap&>(*bucket.column);
+ const auto& cells = assert_cast<const
ColumnString&>(map.get_values());
+ const auto& offsets = map.get_offsets();
+ const size_t begin = row == 0 ? 0 : offsets[row - 1];
+ for (size_t cell = begin; cell < offsets[row]; ++cell) {
+ auto output = encoded.begin_row();
+
RETURN_IF_ERROR(variant_v2::append_v1_storage_cell(cells.get_data_at(cell),
+
output, 0));
+ output.finish();
+ row_paths.push_back(builders.size());
+ }
+ }
+ }
+ while (!pending.empty() && pending.top().first == row) {
+ const size_t index = pending.top().second;
+ pending.pop();
+ const size_t position = positions[index]++;
+ DCHECK(!index_nulls[row]);
+ auto output = encoded.begin_row();
+ RETURN_IF_ERROR(append_materialized_value(values[index],
position, output, 0));
Review Comment:
Fixed. CREATE TABLE now rejects a values index on a VARIANT with declared
Decimal256 paths, direct or inside ARRAY (`VARIANT values index does not
support declared Decimal256 paths`, `IndexDefinition.java:599`), so INSERT can
no longer fail after the DDL succeeded.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ExpressionAnalyzer.java:
##########
@@ -1019,12 +1025,75 @@ public Expression visitMatch(Match match,
ExpressionRewriteContext context) {
right.toSql(), match.toSql(), right.getDataType()));
}
- if (left.getDataType().isVariantType()) {
- left = new Cast(left, right.getDataType());
+ // Only direct roots use leaf-token semantics. Ordinary object
subpaths keep their
+ // existing JSON-to-STRING scalar semantics, including when pruning is
disabled.
+ if (left.getDataType().isVariantType()
+ && (!(left instanceof SlotReference) || ((SlotReference)
left).hasSubColPath()
+ || (currentPlan != null && isScalarVariantMatchInput(left,
currentPlan, new HashMap<>()))
+ || (match.op() != Operator.MATCH_ANY && match.op() !=
Operator.MATCH_ALL))) {
+ left = TypeCoercionUtils.castIfNotSameType(left,
StringType.INSTANCE);
}
+
return match.withChildren(left, right);
}
+ // Alias and set-operation outputs can have neither a subpath nor a source
column.
+ // Prove that every producer is scalar before using STRING semantics. An
unknown
+ // producer or any actual root keeps VARIANT semantics and must bind an
analyzer.
+ private boolean isScalarVariantMatchInput(Expression input, Plan plan,
Map<Slot, Boolean> scalarCteOutputs) {
+ // Set-operation coercion can cast both root columns to a common
VARIANT
+ // layout. That changes storage properties, not root/subpath lineage.
+ if (input instanceof Cast && input.getDataType().isVariantType()
+ && input.child(0).getDataType().isVariantType()) {
+ return isScalarVariantMatchInput(input.child(0), plan,
scalarCteOutputs);
+ }
+ if (!(input instanceof SlotReference)) {
+ return true;
+ }
+ SlotReference slot = (SlotReference) input;
+ if (slot.hasSubColPath()) {
+ return true;
+ }
+ if (plan instanceof LogicalCTEConsumer) {
+ LogicalCTEConsumer consumer = (LogicalCTEConsumer) plan;
+ Slot producerSlot = consumer.getProducerSlot(slot);
+ Boolean scalar = scalarCteOutputs.get(producerSlot);
+ if (scalar != null) {
+ return scalar;
+ }
+ LogicalCTEProducer<?> producer =
getCascadesContext().getStatementContext()
+ .getCteProducerByCteId(consumer.getCteId());
+ // Shared CTE chains form a DAG; do not expand the same producer
at every reference.
+ scalar = isScalarVariantMatchInput(producerSlot, producer.child(),
scalarCteOutputs);
+ scalarCteOutputs.put(producerSlot, scalar);
+ return scalar;
+ }
+ if (plan instanceof LogicalSetOperation) {
+ LogicalSetOperation setOperation = (LogicalSetOperation) plan;
+ int index = setOperation.getOutput().indexOf(slot);
+ if (index >= 0) {
+ for (int i = 0; i < setOperation.arity(); i++) {
+ if
(!isScalarVariantMatchInput(setOperation.getRegularChildrenOutputs().get(i).get(index),
+ setOperation.child(i), scalarCteOutputs)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+ for (Expression expression : plan.getExpressions()) {
+ if (expression instanceof Alias && ((Alias)
expression).getExprId().equals(slot.getExprId())) {
+ return isScalarVariantMatchInput(expression.child(0), plan,
scalarCteOutputs);
+ }
+ }
+ for (Plan child : plan.children()) {
+ if (child.getOutputSet().contains(slot)) {
+ return isScalarVariantMatchInput(slot, child,
scalarCteOutputs);
+ }
+ }
+ return false;
Review Comment:
Resolved at this head. With `enable_match_without_inverted_index=true` and a
values-indexed `v` holding `["hello"]` / `["other"]`, `SELECT element MATCH_ANY
'hello' FROM t LATERAL VIEW explode_variant_array(v) e AS element` returns 1 /
0: the generated slot keeps the scalar STRING MATCH.
##########
be/src/storage/compaction/compaction.cpp:
##########
@@ -135,6 +136,135 @@ namespace {
constexpr size_t kSniiCompactionReadAheadBudgetBytes = 64ULL << 20;
+using SniiSourceFileReaders = std::map<std::string,
std::unique_ptr<IndexFileReader>>;
+
+Result<IndexFileReader*> open_snii_compaction_source(const io::FileSystemSPtr&
fs,
+ const std::string& prefix,
+ const
InvertedIndexFileInfo& file_info,
+ int64_t tablet_id,
+ SniiSourceFileReaders&
readers) {
+ auto it = readers.find(prefix);
+ if (it == readers.end()) {
+ auto reader = std::make_unique<IndexFileReader>(
+ fs, prefix, InvertedIndexStorageFormatPB::SNII, file_info,
tablet_id);
+ const auto status =
reader->init(config::inverted_index_read_buffer_size);
+ if (status.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>()) {
+ return ResultError(Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source SNII index file is unavailable"));
+ }
+ // Missing postings allow rebuild; actual IO failures and corrupt
containers abort.
+ RETURN_IF_ERROR_RESULT(status);
+ DBUG_EXECUTE_IF("Compaction::snii_eligibility_reader_initialized",
DBUG_RUN_CALLBACK());
+ it = readers.emplace(prefix, std::move(reader)).first;
+ }
+ return it->second.get();
+}
+
+Status validate_snii_merge_source(
+ const snii::reader::LogicalIndexReader& reader, const TabletIndex&
source_meta,
+ const TabletIndex& destination, size_t ordinal,
+ std::optional<snii::compaction::SniiCompactionEligibility>&
eligibility) {
+ if (!eligibility.has_value()) {
+ const std::array<snii::compaction::PlainT2CompactionSource, 1> source
= {
+ snii::compaction::PlainT2CompactionSource {.reader =
std::cref(reader),
+ .index_meta =
std::cref(source_meta)}};
+ snii::compaction::SniiCompactionEligibility result;
+
RETURN_IF_ERROR(snii::compaction::validate_snii_compaction_eligibility(source,
destination,
+
&result));
+ eligibility = result;
+ return Status::OK();
+ }
+ if (source_meta.index_id() != destination.index_id() ||
+ source_meta.get_index_suffix() != destination.get_index_suffix() ||
+ source_meta.properties() != destination.properties()) {
+ return Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source SNII index identity or properties differ from
destination");
+ }
+ return snii::compaction::validate_snii_source_eligibility(reader, ordinal,
*eligibility);
+}
+
+Result<uint64_t> check_snii_merge_sources(const std::vector<RowsetSharedPtr>&
rowsets,
+ const TabletIndex& destination,
int64_t tablet_id,
+ SniiSourceFileReaders& readers) {
+ size_t segment_count = 0;
+ for (const auto& rowset : rowsets) {
+ segment_count += rowset->num_segments();
+ }
+ if (segment_count == 0 ||
+ segment_count >
+ kSniiCompactionReadAheadBudgetBytes /
+
snii::compaction::SniiPlainT2MergePlan::kMinReadAheadBudgetPerSource) {
+ return ResultError(Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source index unavailable or read-ahead budget gate failed"));
+ }
+ const int32_t uid = destination.col_unique_ids().front();
+ std::optional<snii::compaction::SniiCompactionEligibility> eligibility;
+ size_t ordinal = 0;
+ // Guaranteed merge allocations: read-ahead plus one ubiquitous term's
docids/frequencies
+ // and the NULL docids from each source. Positions retain the existing
streamed accounting.
+ uint64_t estimated_bytes = kSniiCompactionReadAheadBudgetBytes;
+ for (const auto& rowset : rowsets) {
+ if
(static_cast<BetaRowset*>(rowset.get())->is_skip_index_compaction(uid)) {
+ return ResultError(Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source SNII index compaction is disabled"));
+ }
+ const auto source_metas =
rowset->tablet_schema()->inverted_indexs(uid);
+ const auto source_it = std::ranges::find_if(source_metas, [&](const
TabletIndex* index) {
+ return index->index_id() == destination.index_id();
+ });
+ const auto fs = rowset->rowset_meta()->fs();
+ if (source_it == source_metas.end() || fs == nullptr) {
+ return ResultError(Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source SNII index metadata or file system is
unavailable"));
+ }
+ for (auto segment : rowset->segments()) {
+ const auto path = segment.path();
+ if (!path.has_value()) {
+ return ResultError(Status::Error<INVERTED_INDEX_NOT_SUPPORTED>(
+ "source SNII segment path is unavailable"));
+ }
+ const std::string prefix
{InvertedIndexDescriptor::get_index_file_path_prefix(*path)};
+ const auto file_reader = open_snii_compaction_source(
+ fs, prefix, segment.inverted_index_file_info(), tablet_id,
readers);
+ if (!file_reader.has_value()) {
+ return ResultError(file_reader.error());
+ }
+ auto source =
+ (*file_reader)
+ ->open_snii_index(*source_it, nullptr,
+
snii::reader::LogicalIndexOpenMode::kCompaction);
+ if (!source.has_value() &&
+ source.error().is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>())
{
Review Comment:
Stale. Native merge eligibility now uses master's generic SNII check: a
source whose logical index cannot be opened (including
`INVERTED_INDEX_SNII_NOT_FOUND`) marks that index ineligible, and it is rebuilt
from the raw column (`compaction.cpp:1759`, `:1803`).
`construct_index_compaction_columns()` returns `void`, so this cannot abort
compaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
