diff options
Diffstat (limited to 'src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp')
| -rw-r--r-- | src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp | 594 |
1 files changed, 131 insertions, 463 deletions
diff --git a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp index d38f5992eaf..272d7a1a80c 100644 --- a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp +++ b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp @@ -29,7 +29,6 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery -#include "mongo/db/pipeline/document_source_sequential_document_cache.h" #include <algorithm> #include <iterator> @@ -56,7 +55,6 @@ #include "mongo/db/pipeline/document_source_sample.h" #include "mongo/db/pipeline/document_source_single_document_transformation.h" #include "mongo/db/pipeline/document_source_sort.h" -#include "mongo/db/pipeline/document_source_streaming_group.h" #include "mongo/db/pipeline/expression_context.h" #include "mongo/db/pipeline/lite_parsed_document_source.h" #include "mongo/db/query/query_planner_common.h" @@ -232,7 +230,6 @@ boost::intrusive_ptr<DocumentSourceGroup> createBucketGroupForReorder( void optimizePrefix(Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { auto prefix = Pipeline::SourceContainer(container->begin(), itr); Pipeline::optimizeContainer(&prefix); - Pipeline::optimizeEachStage(&prefix); container->erase(container->begin(), itr); container->splice(itr, prefix); } @@ -246,39 +243,9 @@ DocumentSourceInternalUnpackBucket::DocumentSourceInternalUnpackBucket( bool assumeNoMixedSchemaData) : DocumentSource(kStageNameInternal, expCtx), _assumeNoMixedSchemaData(assumeNoMixedSchemaData), - _bucketUnpacker(std::move(bucketUnpacker)), _bucketMaxSpanSeconds{bucketMaxSpanSeconds} {} -DocumentSourceInternalUnpackBucket::DocumentSourceInternalUnpackBucket( - const boost::intrusive_ptr<ExpressionContext>& expCtx, - BucketUnpacker bucketUnpacker, - int bucketMaxSpanSeconds, - const boost::optional<BSONObj>& eventFilterBson, - const boost::optional<BSONObj>& wholeBucketFilterBson, - bool assumeNoMixedSchemaData) - : DocumentSourceInternalUnpackBucket( - expCtx, std::move(bucketUnpacker), bucketMaxSpanSeconds, assumeNoMixedSchemaData) { - if (eventFilterBson) { - _eventFilterBson = eventFilterBson->getOwned(); - _eventFilter = - uassertStatusOK(MatchExpressionParser::parse(_eventFilterBson, - pExpCtx, - ExtensionsCallbackNoop(), - Pipeline::kAllowedMatcherFeatures)); - _eventFilterDeps = {}; - _eventFilter->addDependencies(&_eventFilterDeps); - } - if (wholeBucketFilterBson) { - _wholeBucketFilterBson = wholeBucketFilterBson->getOwned(); - _wholeBucketFilter = - uassertStatusOK(MatchExpressionParser::parse(_wholeBucketFilterBson, - pExpCtx, - ExtensionsCallbackNoop(), - Pipeline::kAllowedMatcherFeatures)); - } -} - boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createFromBsonInternal( BSONElement specElem, const boost::intrusive_ptr<ExpressionContext>& expCtx) { uassert(5346500, @@ -288,20 +255,14 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF // If neither "include" nor "exclude" is specified, the default is "exclude": [] and // if that's the case, no field will be added to 'bucketSpec.fieldSet' in the for-loop below. + BucketUnpacker::Behavior unpackerBehavior = BucketUnpacker::Behavior::kExclude; BucketSpec bucketSpec; - // Use extended-range support if any individual collection requires it, even if 'specElem' - // doesn't mention this flag. - if (expCtx->getRequiresTimeseriesExtendedRangeSupport()) { - bucketSpec.setUsesExtendedRange(true); - } auto hasIncludeExclude = false; auto hasTimeField = false; auto hasBucketMaxSpanSeconds = false; auto bucketMaxSpanSeconds = 0; auto assumeClean = false; std::vector<std::string> computedMetaProjFields; - boost::optional<BSONObj> eventFilterBson; - boost::optional<BSONObj> wholeBucketFilterBson; for (auto&& elem : specElem.embeddedObject()) { auto fieldName = elem.fieldNameStringData(); if (fieldName == kInclude || fieldName == kExclude) { @@ -326,8 +287,8 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF field.find('.') == std::string::npos); bucketSpec.addIncludeExcludeField(field); } - bucketSpec.setBehavior(fieldName == kInclude ? BucketSpec::Behavior::kInclude - : BucketSpec::Behavior::kExclude); + unpackerBehavior = fieldName == kInclude ? BucketUnpacker::Behavior::kInclude + : BucketUnpacker::Behavior::kExclude; hasIncludeExclude = true; } else if (fieldName == kAssumeNoMixedSchemaData) { uassert(6067202, @@ -389,24 +350,6 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF << " field must be a bool, got: " << elem.type(), elem.type() == BSONType::Bool); bucketSpec.includeMaxTimeAsMetadata = elem.boolean(); - } else if (fieldName == kUsesExtendedRange) { - uassert(6646901, - str::stream() << kUsesExtendedRange - << " field must be a bool, got: " << elem.type(), - elem.type() == BSONType::Bool); - bucketSpec.setUsesExtendedRange(elem.boolean()); - } else if (fieldName == kEventFilter) { - uassert(7026902, - str::stream() << kEventFilter - << " field must be an object, got: " << elem.type(), - elem.type() == BSONType::Object); - eventFilterBson = elem.Obj(); - } else if (fieldName == kWholeBucketFilter) { - uassert(7026903, - str::stream() << kWholeBucketFilter - << " field must be an object, got: " << elem.type(), - elem.type() == BSONType::Object); - wholeBucketFilterBson = elem.Obj(); } else { uasserted(5346506, str::stream() @@ -421,12 +364,11 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF "The $_internalUnpackBucket stage requires a bucketMaxSpanSeconds parameter", hasBucketMaxSpanSeconds); - return make_intrusive<DocumentSourceInternalUnpackBucket>(expCtx, - BucketUnpacker{std::move(bucketSpec)}, - bucketMaxSpanSeconds, - eventFilterBson, - wholeBucketFilterBson, - assumeClean); + return make_intrusive<DocumentSourceInternalUnpackBucket>( + expCtx, + BucketUnpacker{std::move(bucketSpec), unpackerBehavior}, + bucketMaxSpanSeconds, + assumeClean); } boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createFromBsonExternal( @@ -472,49 +414,39 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF hasTimeField); return make_intrusive<DocumentSourceInternalUnpackBucket>( - expCtx, BucketUnpacker{std::move(bucketSpec)}, 3600, assumeClean); + expCtx, + BucketUnpacker{std::move(bucketSpec), BucketUnpacker::Behavior::kExclude}, + 3600, + assumeClean); } -void DocumentSourceInternalUnpackBucket::serializeToArray(std::vector<Value>& array, - const SerializationOptions& opts) const { - auto explain = opts.verbosity; - +void DocumentSourceInternalUnpackBucket::serializeToArray( + std::vector<Value>& array, boost::optional<ExplainOptions::Verbosity> explain) const { MutableDocument out; auto behavior = - _bucketUnpacker.behavior() == BucketSpec::Behavior::kInclude ? kInclude : kExclude; + _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kInclude ? kInclude : kExclude; const auto& spec = _bucketUnpacker.bucketSpec(); std::vector<Value> fields; for (auto&& field : spec.fieldSet()) { - fields.emplace_back(opts.serializeFieldPathFromString(field)); + fields.emplace_back(field); } if (((_bucketUnpacker.includeMetaField() && - _bucketUnpacker.behavior() == BucketSpec::Behavior::kInclude) || + _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kInclude) || (!_bucketUnpacker.includeMetaField() && - _bucketUnpacker.behavior() == BucketSpec::Behavior::kExclude && spec.metaField())) && + _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kExclude && spec.metaField())) && std::find(spec.computedMetaProjFields().cbegin(), spec.computedMetaProjFields().cend(), *spec.metaField()) == spec.computedMetaProjFields().cend()) - fields.emplace_back(opts.serializeFieldPathFromString(*spec.metaField())); + fields.emplace_back(*spec.metaField()); out.addField(behavior, Value{std::move(fields)}); - out.addField(timeseries::kTimeFieldName, - Value{opts.serializeFieldPathFromString(spec.timeField())}); + out.addField(timeseries::kTimeFieldName, Value{spec.timeField()}); if (spec.metaField()) { - out.addField(timeseries::kMetaFieldName, - Value{opts.serializeFieldPathFromString(*spec.metaField())}); + out.addField(timeseries::kMetaFieldName, Value{*spec.metaField()}); } - out.addField(kBucketMaxSpanSeconds, opts.serializeLiteral(Value{_bucketMaxSpanSeconds})); + out.addField(kBucketMaxSpanSeconds, Value{_bucketMaxSpanSeconds}); if (_assumeNoMixedSchemaData) - out.addField(kAssumeNoMixedSchemaData, - opts.serializeLiteral(Value(_assumeNoMixedSchemaData))); - - if (spec.usesExtendedRange()) { - // Include this flag so that 'explain' is more helpful. - // But this is not so useful for communicating from one process to another, - // because mongos and/or the primary shard don't know whether any other shard - // has extended-range data. - out.addField(kUsesExtendedRange, opts.serializeLiteral(Value{true})); - } + out.addField(kAssumeNoMixedSchemaData, Value(_assumeNoMixedSchemaData)); if (!spec.computedMetaProjFields().empty()) out.addField("computedMetaProjFields", Value{[&] { @@ -522,97 +454,51 @@ void DocumentSourceInternalUnpackBucket::serializeToArray(std::vector<Value>& ar std::transform(spec.computedMetaProjFields().cbegin(), spec.computedMetaProjFields().cend(), std::back_inserter(compFields), - [opts](auto&& projString) { - return Value{ - opts.serializeFieldPathFromString(projString)}; - }); + [](auto&& projString) { return Value{projString}; }); return compFields; }()}); if (_bucketUnpacker.includeMinTimeAsMetadata()) { - out.addField(kIncludeMinTimeAsMetadata, - opts.serializeLiteral(Value{_bucketUnpacker.includeMinTimeAsMetadata()})); + out.addField(kIncludeMinTimeAsMetadata, Value{_bucketUnpacker.includeMinTimeAsMetadata()}); } if (_bucketUnpacker.includeMaxTimeAsMetadata()) { - out.addField(kIncludeMaxTimeAsMetadata, - opts.serializeLiteral(Value{_bucketUnpacker.includeMaxTimeAsMetadata()})); - } - - if (_wholeBucketFilter) { - out.addField(kWholeBucketFilter, Value{_wholeBucketFilter->serialize(opts)}); - } - if (_eventFilter) { - out.addField(kEventFilter, Value{_eventFilter->serialize(opts)}); + out.addField(kIncludeMaxTimeAsMetadata, Value{_bucketUnpacker.includeMaxTimeAsMetadata()}); } if (!explain) { array.push_back(Value(DOC(getSourceName() << out.freeze()))); if (_sampleSize) { auto sampleSrc = DocumentSourceSample::create(pExpCtx, *_sampleSize); - sampleSrc->serializeToArray(array, opts); + sampleSrc->serializeToArray(array); } } else { if (_sampleSize) { - out.addField("sample", - opts.serializeLiteral(Value{static_cast<long long>(*_sampleSize)})); - out.addField("bucketMaxCount", opts.serializeLiteral(Value{_bucketMaxCount})); + out.addField("sample", Value{static_cast<long long>(*_sampleSize)}); + out.addField("bucketMaxCount", Value{_bucketMaxCount}); } array.push_back(Value(DOC(getSourceName() << out.freeze()))); } } -boost::optional<Document> DocumentSourceInternalUnpackBucket::getNextMatchingMeasure() { - while (_bucketUnpacker.hasNext()) { - if (_eventFilter) { - if (_unpackToBson) { - auto measure = _bucketUnpacker.getNextBson(); - if (_bucketUnpacker.bucketMatchedQuery() || _eventFilter->matchesBSON(measure)) { - return Document(measure); - } - } else { - auto measure = _bucketUnpacker.getNext(); - // MatchExpression only takes BSON documents, so we have to make one. As an - // optimization, only serialize the fields we need to do the match. - BSONObj measureBson = _eventFilterDeps.needWholeDocument - ? measure.toBson() - : document_path_support::documentToBsonWithPaths(measure, - _eventFilterDeps.fields); - if (_bucketUnpacker.bucketMatchedQuery() || - _eventFilter->matchesBSON(measureBson)) { - return measure; - } - } - } else { - return _bucketUnpacker.getNext(); - } - } - return {}; -} - DocumentSource::GetNextResult DocumentSourceInternalUnpackBucket::doGetNext() { tassert(5521502, "calling doGetNext() when '_sampleSize' is set is disallowed", !_sampleSize); // Otherwise, fallback to unpacking every measurement in all buckets until the child stage is // exhausted. - if (auto measure = getNextMatchingMeasure()) { - return GetNextResult(std::move(*measure)); + if (_bucketUnpacker.hasNext()) { + return _bucketUnpacker.getNext(); } auto nextResult = pSource->getNext(); - while (nextResult.isAdvanced()) { + if (nextResult.isAdvanced()) { auto bucket = nextResult.getDocument().toBson(); - auto bucketMatchedQuery = _wholeBucketFilter && _wholeBucketFilter->matchesBSON(bucket); - _bucketUnpacker.reset(std::move(bucket), bucketMatchedQuery); - + _bucketUnpacker.reset(std::move(bucket)); uassert(5346509, str::stream() << "A bucket with _id " << _bucketUnpacker.bucket()[timeseries::kBucketIdFieldName].toString() << " contains an empty data region", _bucketUnpacker.hasNext()); - if (auto measure = getNextMatchingMeasure()) { - return GetNextResult(std::move(*measure)); - } - nextResult = pSource->getNext(); + return _bucketUnpacker.getNext(); } return nextResult; @@ -624,7 +510,7 @@ bool DocumentSourceInternalUnpackBucket::pushDownComputedMetaProjection( if (std::next(itr) == container->end()) { return nextStageWasRemoved; } - if (!_bucketUnpacker.getMetaField() || !_bucketUnpacker.includeMetaField()) { + if (!_bucketUnpacker.bucketSpec().metaField()) { return nextStageWasRemoved; } @@ -676,8 +562,9 @@ void DocumentSourceInternalUnpackBucket::internalizeProject(const BSONObj& proje // Update '_bucketUnpacker' state with the new fields and behavior. auto spec = _bucketUnpacker.bucketSpec(); spec.setFieldSet(fields); - spec.setBehavior(isInclusion ? BucketSpec::Behavior::kInclude : BucketSpec::Behavior::kExclude); - _bucketUnpacker.setBucketSpec(std::move(spec)); + _bucketUnpacker.setBucketSpecAndBehavior(std::move(spec), + isInclusion ? BucketUnpacker::Behavior::kInclude + : BucketUnpacker::Behavior::kExclude); } std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProjectToInternalize( @@ -689,8 +576,7 @@ std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProje // Check for a viable inclusion $project after the $_internalUnpackBucket. auto [existingProj, isInclusion] = getIncludeExcludeProjectAndType(std::next(itr)->get()); - if (!_eventFilter && isInclusion && !existingProj.isEmpty() && - canInternalizeProjectObj(existingProj)) { + if (isInclusion && !existingProj.isEmpty() && canInternalizeProjectObj(existingProj)) { container->erase(std::next(itr)); return {existingProj, isInclusion}; } @@ -698,7 +584,8 @@ std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProje // Attempt to get an inclusion $project representing the root-level dependencies of the pipeline // after the $_internalUnpackBucket. If this $project is not empty, then the dependency set was // finite. - auto deps = getRestPipelineDependencies(itr, container, true /* includeEventFilter */); + Pipeline::SourceContainer restOfPipeline(std::next(itr), container->end()); + auto deps = Pipeline::getDependenciesForContainer(pExpCtx, restOfPipeline, boost::none); if (auto dependencyProj = deps.toProjectionWithoutMetadata(DepsTracker::TruncateToRootLevel::yes); !dependencyProj.isEmpty()) { @@ -706,7 +593,7 @@ std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProje } // Check for a viable exclusion $project after the $_internalUnpackBucket. - if (!_eventFilter && !existingProj.isEmpty() && canInternalizeProjectObj(existingProj)) { + if (!existingProj.isEmpty() && canInternalizeProjectObj(existingProj)) { container->erase(std::next(itr)); return {existingProj, isInclusion}; } @@ -714,7 +601,8 @@ std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProje return {BSONObj{}, false}; } -BucketSpec::BucketPredicate DocumentSourceInternalUnpackBucket::createPredicatesOnBucketLevelField( +std::unique_ptr<MatchExpression> +DocumentSourceInternalUnpackBucket::createPredicatesOnBucketLevelField( const MatchExpression* matchExpr) const { return BucketSpec::createPredicatesOnBucketLevelField( matchExpr, @@ -743,102 +631,58 @@ std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractProjectForPu std::pair<bool, Pipeline::SourceContainer::iterator> DocumentSourceInternalUnpackBucket::rewriteGroupByMinMax(Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { - // The computed min/max for each bucket uses the default collation. If the collation of the - // query doesn't match the default we cannot rely on the computed values as they might differ - // (e.g. numeric and lexicographic collations compare "5" and "10" in opposite order). - // NB: Unfortuntealy, this means we have to forgo the optimization even if the source field is - // numeric and not affected by the collation as we cannot know the data type until runtime. - if (pExpCtx->collationMatchesDefault == ExpressionContext::CollationMatchesDefault::kNo) { - return {}; - } - const auto* groupPtr = dynamic_cast<DocumentSourceGroup*>(std::next(itr)->get()); if (groupPtr == nullptr) { return {}; } - if (!_bucketUnpacker.bucketSpec().metaField()) { - return {}; - } - const auto& metaField = *_bucketUnpacker.bucketSpec().metaField(); - const auto& idFields = groupPtr->getIdFields(); - - // Currently, we only support simple group key. TODO: SERVER-68811. Allow rewrites of object - // group key if all its fields depend on the metaField only. - if (idFields.size() != 1) { + if (idFields.size() != 1 || !_bucketUnpacker.bucketSpec().metaField().has_value()) { return {}; } const auto& exprId = idFields.cbegin()->second; const auto* exprIdPath = dynamic_cast<const ExpressionFieldPath*>(exprId.get()); - - // Currently, we only support group key expression of the form {_id : "<path>"}. - // TODO: SERVER-68811. Allow rewrites if expression is constant. if (exprIdPath == nullptr) { return {}; } const auto& idPath = exprIdPath->getFieldPath(); - // {The path must be at or under the metaField for this re-write to be correct (and the zero - // component is always CURRENT).} - if (idPath.getPathLength() < 2 || idPath.getFieldName(1) != metaField) { + if (idPath.getPathLength() < 2 || + idPath.getFieldName(1) != _bucketUnpacker.bucketSpec().metaField().get()) { return {}; } - std::vector<AccumulationStatement> accumulationStatementsBucket; + bool suitable = true; + std::vector<AccumulationStatement> accumulationStatements; for (const AccumulationStatement& stmt : groupPtr->getAccumulatedFields()) { - const auto& op = stmt.expr.name; - // If _any_ of the accumulators aren't $min/$max we won't perform the re-write (some other - // accs might be re-writable in terms of the bucket controls, we just haven't invested into - // implementing them). - if (op != "$min" && op != "$max") { - return {}; - } - - const auto* exprArgPath = - dynamic_cast<const ExpressionFieldPath*>(stmt.expr.argument.get()); - - // This is either a const or a compound expression. While some such expressions (e.g: {$min: - // {$add: - // ['$a', 2]}}) could be re-written in terms of the min/max on the control fields, in - // general we cannot do it (e.g. {$min: {$add: ['$a', '$b']}}), so we block the re-write. - if (!exprArgPath) { - return {}; - } - - // Path can have a single component if it's using $$CURRENT or a similar variable. We don't - // support these. - const auto& path = exprArgPath->getFieldPath(); - if (path.getPathLength() <= 1) { - return {}; - } - const auto& accFieldName = path.getFieldName(1); + const auto op = stmt.expr.name; + const bool isMin = op == "$min"; + const bool isMax = op == "$max"; - // Rewrite not valid for the timeField because control.min.time contains a rounded-down time - // and not the actual min time of events in the bucket. - if (accFieldName == _bucketUnpacker.bucketSpec().timeField()) { - return {}; + // Rewrite is valid only for min and max aggregates. + if (!isMin && !isMax) { + suitable = false; + break; } - // Build the paths for the bucket-level fields. - std::ostringstream os; - if (accFieldName == metaField) { - // Update aggregates to reference the meta field. - os << timeseries::kBucketMetaFieldName; - - for (size_t index = 2; index < path.getPathLength(); index++) { - os << "." << path.getFieldName(index); + const auto* exprArg = stmt.expr.argument.get(); + if (const auto* exprArgPath = dynamic_cast<const ExpressionFieldPath*>(exprArg)) { + const auto& path = exprArgPath->getFieldPath(); + if (path.getPathLength() <= 1 || + path.getFieldName(1) == _bucketUnpacker.bucketSpec().timeField()) { + // Rewrite not valid for time field. We want to eliminate the bucket + // unpack stage here. + suitable = false; + break; } - } else { + // Update aggregates to reference the control field. - const auto op = stmt.expr.name; - if (op == "$min") { + std::ostringstream os; + if (isMin) { os << timeseries::kControlMinFieldNamePrefix; - } else if (op == "$max") { - os << timeseries::kControlMaxFieldNamePrefix; } else { - MONGO_UNREACHABLE; + os << timeseries::kControlMaxFieldNamePrefix; } for (size_t index = 1; index < path.getPathLength(); index++) { @@ -847,55 +691,45 @@ DocumentSourceInternalUnpackBucket::rewriteGroupByMinMax(Pipeline::SourceContain } os << path.getFieldName(index); } - } - // Re-create the accumulator using the bucket-level paths. - const auto& newExpr = ExpressionFieldPath::createPathFromString( - pExpCtx.get(), os.str(), pExpCtx->variablesParseState); - - AccumulationExpression accExpr = stmt.expr; - accExpr.argument = newExpr; - accumulationStatementsBucket.emplace_back(stmt.fieldName, std::move(accExpr)); - } + const auto& newExpr = ExpressionFieldPath::createPathFromString( + pExpCtx.get(), os.str(), pExpCtx->variablesParseState); - // Re-create the group key using the bucket-level path. - std::ostringstream os; - os << timeseries::kBucketMetaFieldName; - for (size_t index = 2; index < idPath.getPathLength(); index++) { - os << "." << idPath.getFieldName(index); + AccumulationExpression accExpr = stmt.expr; + accExpr.argument = newExpr; + accumulationStatements.emplace_back(stmt.fieldName, std::move(accExpr)); + } } - auto exprIdBucket = ExpressionFieldPath::createPathFromString( - pExpCtx.get(), os.str(), pExpCtx->variablesParseState); - boost::intrusive_ptr<Expression> rewrittenIdExpression; - const auto& idFieldNames = groupPtr->getIdFieldNames(); - if (idFieldNames.empty()) { - rewrittenIdExpression = exprIdBucket; - } else { - // idFieldNames can only have size 1 here since we only support simple group key. - // TODO: SERVER-68811. Allow rewrites of object group key if all its fields depend on the - // metaField only. - rewrittenIdExpression = - ExpressionObject::create(pExpCtx.get(), {{idFieldNames[0], exprIdBucket}}); - } + if (suitable) { + std::ostringstream os; + os << timeseries::kBucketMetaFieldName; + for (size_t index = 2; index < idPath.getPathLength(); index++) { + os << "." << idPath.getFieldName(index); + } + auto exprId1 = ExpressionFieldPath::createPathFromString( + pExpCtx.get(), os.str(), pExpCtx->variablesParseState); - auto newGroup = DocumentSourceGroup::create(pExpCtx, - std::move(rewrittenIdExpression), - std::move(accumulationStatementsBucket), - groupPtr->getMaxMemoryUsageBytes()); + auto newGroup = DocumentSourceGroup::create(pExpCtx, + std::move(exprId1), + std::move(accumulationStatements), + groupPtr->getMaxMemoryUsageBytes()); - // Replace the current stage (DocumentSourceInternalUnpackBucket) and the following group stage - // with the new group. - container->erase(std::next(itr)); - *itr = std::move(newGroup); + // Erase current stage and following group stage, and replace with updated + // group. + container->erase(std::next(itr)); + *itr = std::move(newGroup); - if (itr == container->begin()) { - // Optimize the new group stage. - return {true, itr}; - } else { - // Give chance to the previous stage to optimize against the new group stage. - return {true, std::prev(itr)}; + if (itr == container->begin()) { + // Optimize group stage. + return {true, itr}; + } else { + // Give chance of the previous stage to optimize against group stage. + return {true, std::prev(itr)}; + } } + + return {}; } bool DocumentSourceInternalUnpackBucket::haveComputedMetaField() const { @@ -904,68 +738,6 @@ bool DocumentSourceInternalUnpackBucket::haveComputedMetaField() const { _bucketUnpacker.bucketSpec().metaField().get()); } -bool DocumentSourceInternalUnpackBucket::enableStreamingGroupIfPossible( - Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { - // skip unpack stage - itr = std::next(itr); - - FieldPath timeField = _bucketUnpacker.bucketSpec().timeField(); - DocumentSourceGroup* groupStage = nullptr; - bool isSortedOnTime = false; - for (; itr != container->end(); ++itr) { - if (auto groupStagePtr = dynamic_cast<DocumentSourceGroup*>(itr->get())) { - groupStage = groupStagePtr; - break; - } - if (auto sortStagePtr = dynamic_cast<DocumentSourceSort*>(itr->get())) { - isSortedOnTime = sortStagePtr->getSortKeyPattern().front().fieldPath == timeField; - } else if (!itr->get()->constraints().preservesOrderAndMetadata) { - // If this is after the sort, the sort is invalidated. If it's before the sort, there's - // no harm in keeping the boolean false. - isSortedOnTime = false; - } - // We modify time field, so we can't proceed with optimization. It may be possible to - // proceed in some cases if the modification happens before the sort, but we won't worry - // about or bother with those - in large part because it is risky that it will change the - // type away from a date into something with more difficult/subtle semantics. - if (itr->get()->getModifiedPaths().canModify(timeField)) { - return false; - } - } - - if (groupStage == nullptr || !isSortedOnTime) { - return false; - } - - const auto& idFields = groupStage->getMutableIdFields(); - std::vector<size_t> monotonicIdFields; - for (size_t i = 0; i < idFields.size(); ++i) { - // To enable streaming, we need id field expression to be clustered, so that all documents - // with the same value of this id field are in a single continious cluster. However this - // property is hard to check for, so we check for monotonicity instead, which is stronger. - idFields[i]->optimize(); // We optimize here to make use of constant folding. - auto monotonicState = idFields[i]->getMonotonicState(timeField); - - // We don't add monotonic::State::Constant id fields, because they are useless when - // determining if a group batch is finished. - if (monotonicState == monotonic::State::Increasing || - monotonicState == monotonic::State::Decreasing) { - monotonicIdFields.push_back(i); - } - } - if (monotonicIdFields.empty()) { - return false; - } - - *itr = - DocumentSourceStreamingGroup::create(pExpCtx, - groupStage->getIdExpression(), - std::move(monotonicIdFields), - std::move(groupStage->getMutableAccumulatedFields()), - groupStage->getMaxMemoryUsageBytes()); - return true; -} - template <TopBottomSense sense, bool single> bool extractFromAcc(const AccumulatorN* acc, const boost::intrusive_ptr<Expression>& init, @@ -1208,55 +980,6 @@ bool DocumentSourceInternalUnpackBucket::optimizeLastpoint(Pipeline::SourceConta tryInsertBucketLevelSortAndGroup(AccumulatorDocumentsNeeded::kLastDocument); } - -bool findSequentialDocumentCache(Pipeline::SourceContainer::iterator start, - Pipeline::SourceContainer::iterator end) { - while (start != end && !dynamic_cast<DocumentSourceSequentialDocumentCache*>(start->get())) { - start = std::next(start); - } - return start != end; -} - -Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::optimizeAtRestOfPipeline( - Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { - if (itr == container->end()) { - return itr; - } - - invariant(*itr == this); - Pipeline::SourceContainer::iterator unpackBucket = itr; - - itr = std::next(itr); - - try { - while (itr != container->end()) { - if (itr == unpackBucket) { - itr = std::next(itr); - if (itr == container->end()) - break; - } - itr = (*itr).get()->optimizeAt(itr, container); - } - } catch (DBException& ex) { - ex.addContext("Failed to optimize pipeline"); - throw; - } - - return itr; -} - -DepsTracker DocumentSourceInternalUnpackBucket::getRestPipelineDependencies( - Pipeline::SourceContainer::iterator itr, - Pipeline::SourceContainer* container, - bool includeEventFilter) const { - auto deps = Pipeline::getDependenciesForContainer( - pExpCtx, Pipeline::SourceContainer{std::next(itr), container->end()}, boost::none); - if (_eventFilter && includeEventFilter) { - _eventFilter->addDependencies(&deps); - } - return deps; -} - Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimizeAt( Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { invariant(*itr == this); @@ -1270,8 +993,7 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi bool haveComputedMetaField = this->haveComputedMetaField(); // Before any other rewrites for the current stage, consider reordering with $sort. - if (auto sortPtr = dynamic_cast<DocumentSourceSort*>(std::next(itr)->get()); - sortPtr && !_eventFilter) { + if (auto sortPtr = dynamic_cast<DocumentSourceSort*>(std::next(itr)->get())) { if (auto metaField = _bucketUnpacker.bucketSpec().metaField(); metaField && !haveComputedMetaField) { if (checkMetadataSortReorder(sortPtr->getSortKeyPattern(), metaField.get())) { @@ -1302,8 +1024,7 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi } // Attempt to push geoNear on the metaField past $_internalUnpackBucket. - if (auto nextNear = dynamic_cast<DocumentSourceGeoNear*>(std::next(itr)->get()); - nextNear && !_eventFilter) { + if (auto nextNear = dynamic_cast<DocumentSourceGeoNear*>(std::next(itr)->get())) { // Currently we only support geo indexes on the meta field, and we enforce this by // requiring the key field to be set so we can check before we try to look up indexes. auto keyField = nextNear->getKeyField(); @@ -1337,34 +1058,23 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi } } - // OptimizeAt the pipeline after this stage to merge $match stages and push them forward. + // Optimize the pipeline after this stage to merge $match stages and push them forward, and to + // take advantage of $expr rewrite optimizations. if (!_optimizedEndOfPipeline) { _optimizedEndOfPipeline = true; if (std::next(itr) == container->end()) { return container->end(); } - - auto cacheFound = findSequentialDocumentCache(itr, container->end()); - if (cacheFound) { - // We want to call optimizeAt() on the rest of the pipeline first, and exit this - // function since any calls to optimize() will interfere with the - // sequentialDocumentCache's ability to properly place itself or abandon. - return DocumentSourceInternalUnpackBucket::optimizeAtRestOfPipeline(itr, container); - } else { - if (auto nextStage = dynamic_cast<DocumentSourceGeoNear*>(std::next(itr)->get())) { - // If the end of the pipeline starts with a $geoNear stage, make sure it gets - // optimized in a context where it knows there are other stages before it. It will - // split itself up into separate $match and $sort stages. But it doesn't split - // itself up when it's the first stage, because it expects to use a special - // DocumentSouceGeoNearCursor plan. - nextStage->optimizeAt(std::next(itr), container); - } - // We want to optimize the rest of the pipeline to ensure the stages are in their - // optimal position and expressions have been optimized to allow for certain rewrites. - Pipeline::optimizeEndOfPipeline(itr, container); + if (auto nextStage = dynamic_cast<DocumentSourceGeoNear*>(std::next(itr)->get())) { + // If the end of the pipeline starts with a $geoNear stage, make sure it gets optimized + // in a context where it knows there are other stages before it. It will split itself + // up into separate $match and $sort stages. But it doesn't split itself up when it's + // the first stage, because it expects to use a special DocumentSouceGeoNearCursor plan. + nextStage->optimizeAt(std::next(itr), container); } + Pipeline::optimizeEndOfPipeline(itr, container); if (std::next(itr) == container->end()) { return container->end(); } else { @@ -1373,8 +1083,7 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi return itr; } } - - if (!_eventFilter) { + { // Check if we can avoid unpacking if we have a group stage with min/max aggregates. auto [success, result] = rewriteGroupByMinMax(itr, container); if (success) { @@ -1385,12 +1094,13 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi { // Check if the rest of the pipeline needs any fields. For example we might only be // interested in $count. - auto deps = getRestPipelineDependencies(itr, container, true /* includeEventFilter */); + auto deps = Pipeline::getDependenciesForContainer( + pExpCtx, Pipeline::SourceContainer{std::next(itr), container->end()}, boost::none); if (deps.hasNoRequirements()) { - _bucketUnpacker.setBucketSpec({_bucketUnpacker.bucketSpec().timeField(), - _bucketUnpacker.bucketSpec().metaField(), - {}, - BucketSpec::Behavior::kInclude}); + _bucketUnpacker.setBucketSpecAndBehavior({_bucketUnpacker.bucketSpec().timeField(), + _bucketUnpacker.bucketSpec().metaField(), + {}}, + BucketUnpacker::Behavior::kInclude); // Keep going for next optimization. } @@ -1407,71 +1117,31 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi // Attempt to optimize last-point type queries. if (feature_flags::gfeatureFlagLastPointQuery.isEnabled( serverGlobalParams.featureCompatibility) && - !_triedLastpointRewrite && !_eventFilter && optimizeLastpoint(itr, container)) { + !_triedLastpointRewrite && optimizeLastpoint(itr, container)) { _triedLastpointRewrite = true; // If we are able to rewrite the aggregation, give the resulting pipeline a chance to // perform further optimizations. return container->begin(); }; - // Attempt to map predicates on bucketed fields to the predicates on the control field. - if (auto nextMatch = dynamic_cast<DocumentSourceMatch*>(std::next(itr)->get())) { + // Attempt to map predicates on bucketed fields to predicates on the control field. + if (auto nextMatch = dynamic_cast<DocumentSourceMatch*>(std::next(itr)->get()); + nextMatch && !_triedBucketLevelFieldsPredicatesPushdown) { + _triedBucketLevelFieldsPredicatesPushdown = true; - // Merge multiple following $match stages. - auto itrToMatch = std::next(itr); - while (std::next(itrToMatch) != container->end() && - dynamic_cast<DocumentSourceMatch*>(std::next(itrToMatch)->get())) { - nextMatch->doOptimizeAt(itrToMatch, container); - } - - auto predicates = createPredicatesOnBucketLevelField(nextMatch->getMatchExpression()); - - // Try to create a tight bucket predicate to perform bucket level matching. - if (predicates.tightPredicate) { - _wholeBucketFilterBson = predicates.tightPredicate->serialize(); - _wholeBucketFilter = - uassertStatusOK(MatchExpressionParser::parse(_wholeBucketFilterBson, - pExpCtx, - ExtensionsCallbackNoop(), - Pipeline::kAllowedMatcherFeatures)); - _wholeBucketFilter = MatchExpression::optimize(std::move(_wholeBucketFilter)); - } - - // Push the original event predicate into the unpacking stage. - _eventFilterBson = nextMatch->getQuery().getOwned(); - _eventFilter = - uassertStatusOK(MatchExpressionParser::parse(_eventFilterBson, - pExpCtx, - ExtensionsCallbackNoop(), - Pipeline::kAllowedMatcherFeatures)); - _eventFilter = MatchExpression::optimize(std::move(_eventFilter)); - _eventFilterDeps = {}; - _eventFilter->addDependencies(&_eventFilterDeps); - container->erase(std::next(itr)); - - // If the $match is not followed by other stages referencing fields (e.g. $count), we can - // unpack directly to BSON so that data doesn't need to be materialized to Document. - auto deps = getRestPipelineDependencies(itr, container, false /* includeEventFilter */); - if (deps.fields.empty()) { - _unpackToBson = true; - } - - // Create a loose bucket predicate and push it before the unpacking stage. - if (predicates.loosePredicate) { - container->insert( - itr, DocumentSourceMatch::create(predicates.loosePredicate->serialize(), pExpCtx)); + if (auto match = createPredicatesOnBucketLevelField(nextMatch->getMatchExpression())) { + BSONObjBuilder bob; + match->serialize(&bob); + container->insert(itr, DocumentSourceMatch::create(bob.obj(), pExpCtx)); // Give other stages a chance to optimize with the new $match. return std::prev(itr) == container->begin() ? std::prev(itr) : std::prev(std::prev(itr)); } - - // We have removed a $match after this stage, so we try to optimize this stage again. - return itr; } // Attempt to push down a $project on the metaField past $_internalUnpackBucket. - if (!_eventFilter && !haveComputedMetaField) { + if (!haveComputedMetaField) { if (auto [metaProject, deleteRemainder] = extractProjectForPushDown(std::next(itr)->get()); !metaProject.isEmpty()) { container->insert(itr, @@ -1490,7 +1160,7 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi // Attempt to extract computed meta projections from subsequent $project, $addFields, or $set // and push them before the $_internalunpackBucket. - if (!_eventFilter && pushDownComputedMetaProjection(itr, container)) { + if (pushDownComputedMetaProjection(itr, container)) { // We've pushed down and removed a stage after this one. Try to optimize the new stage. return std::prev(itr) == container->begin() ? std::prev(itr) : std::prev(std::prev(itr)); } @@ -1509,8 +1179,6 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi } } - enableStreamingGroupIfPossible(itr, container); - return container->end(); } @@ -1519,8 +1187,8 @@ DocumentSource::GetModPathsReturn DocumentSourceInternalUnpackBucket::getModifie StringMap<std::string> renames; renames.emplace(*_bucketUnpacker.bucketSpec().metaField(), timeseries::kBucketMetaFieldName); - return {GetModPathsReturn::Type::kAllExcept, OrderedPathSet{}, std::move(renames)}; + return {GetModPathsReturn::Type::kAllExcept, std::set<std::string>{}, std::move(renames)}; } - return {GetModPathsReturn::Type::kAllPaths, OrderedPathSet{}, {}}; + return {GetModPathsReturn::Type::kAllPaths, std::set<std::string>{}, {}}; } } // namespace mongo |