Skip to content

Add RowGrouping to fix Row Merging on MultiEngine Queries #447

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pranavbhole merged 14 commits into from
May 9, 2019
Merged

Add RowGrouping to fix Row Merging on MultiEngine Queries #447

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
c9b9ae7
Add test to illustrate MultiEngineQuery issue.
ryankwagner Apr 30, 2019
ea57c50
Add initial addRow new Functionality with RowGrouping. Test illustra…
ryankwagner May 1, 2019
9231e1f
Fix to fit DimDriven cases, scoverage failing min coverage.
ryankwagner May 2, 2019
e2154b4
Coverage reached, executors fixed with new RowGrouping logic.
ryankwagner May 2, 2019
b1a7176
Remove redundant code, clean.
ryankwagner May 3, 2019
3a14798
Remove excess row testing & unnecessary row addition.
ryankwagner May 3, 2019
4d5d4ba
Fix multiEngineQuery to only GroupBy Dim Primary Keys.
ryankwagner May 6, 2019
f9f8d24
Fix dim driven queries
ryankwagner May 6, 2019
a1075a3
Allow for multiple value add to rows
ryankwagner May 7, 2019
b090f19
Merge branch 'master' of https://github.com/yahoo/maha into multiEngi…
ryankwagner May 7, 2019
3292a87
Revert changes pushed into another branch.
ryankwagner May 7, 2019
f923a8a
Revert changes pushed into another branch.
ryankwagner May 7, 2019
6a7e6b3
Add fact and Dim RowList tests to demonstrate changes individually.
ryankwagner May 7, 2019
6a3d7d7
Merge with master & new tests
ryankwagner May 8, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions core/src/main/scala/com/yahoo/maha/core/query/QueryContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import scala.collection.SortedSet
sealed trait QueryContext {
def requestModel: RequestModel
def indexAliasOption: Option[String]
def factGroupByKeys: List[String]
def primaryTableName: String
}

Expand Down Expand Up @@ -59,17 +60,20 @@ case object DimFactOuterGroupByQuery extends QueryType

case class DimQueryContext private[query](dims: SortedSet[DimensionBundle],
requestModel: RequestModel,
indexAliasOption: Option[String],
indexAliasOption: Option[String],
factGroupByKeys: List[String],
queryAttributes: QueryAttributes= QueryAttributes.empty) extends DimensionQueryContext
case class FactQueryContext private[query](factBestCandidate: FactBestCandidate,
requestModel: RequestModel,
indexAliasOption: Option[String],
indexAliasOption: Option[String],
factGroupByKeys: List[String],
queryAttributes: QueryAttributes) extends FactualQueryContext
case class CombinedQueryContext private[query](dims: SortedSet[DimensionBundle],
factBestCandidate: FactBestCandidate,
requestModel: RequestModel,
queryAttributes: QueryAttributes) extends DimensionQueryContext with FactualQueryContext {
val indexAliasOption = None
val factGroupByKeys = List.empty
override def primaryTableName: String = {
if(requestModel.isDimDriven) {
dims.last.dim.name
Expand All @@ -84,6 +88,7 @@ case class DimFactOuterGroupByQueryQueryContext(dims: SortedSet[DimensionBundle]
requestModel: RequestModel,
queryAttributes: QueryAttributes) extends DimensionQueryContext with FactualQueryContext {
override def indexAliasOption: Option[String] = None
override def factGroupByKeys: List[String] = List.empty
override def primaryTableName: String = factBestCandidate.fact.name

lazy val shouldQualifyFactsInPreOuter: Boolean = {
Expand Down Expand Up @@ -143,6 +148,7 @@ class QueryContextBuilder(queryType: QueryType, requestModel: RequestModel) {
var dims : SortedSet[DimensionBundle] = SortedSet.empty
var factBestCandidate : Option[FactBestCandidate] = None
var indexAliasOption : Option[String] = None
var factGroupByKeys : List[String] = List.empty
var queryAttributes : QueryAttributes = QueryAttributes.empty

def addDimTable(dimension: DimensionBundle) = {
Expand Down Expand Up @@ -170,6 +176,13 @@ class QueryContextBuilder(queryType: QueryType, requestModel: RequestModel) {
this
}

def addFactGroupByKeys(factGroupByKeyList: List[String]) = {
require(queryType != DimFactQuery, "dim fact query doesn't use Group By keys")
require(factGroupByKeys.isEmpty, s"fact group by is already defined : factGroupByKeys=${factGroupByKeys.toString()}, cannot set to ${factGroupByKeyList.toString()}")
this.factGroupByKeys = factGroupByKeyList
this
}

def setQueryAttributes(queryAttributes: QueryAttributes) = {
this.queryAttributes = queryAttributes
this
Expand All @@ -179,10 +192,10 @@ class QueryContextBuilder(queryType: QueryType, requestModel: RequestModel) {
queryType match {
case DimOnlyQuery =>
require(dims.nonEmpty, "dim only query should not have dimension empty")
DimQueryContext(dims, requestModel, indexAliasOption, queryAttributes)
DimQueryContext(dims, requestModel, indexAliasOption, factGroupByKeys, queryAttributes)
case FactOnlyQuery =>
require(factBestCandidate.isDefined, "fact only query should have fact defined")
FactQueryContext(factBestCandidate.get, requestModel, indexAliasOption, queryAttributes)
FactQueryContext(factBestCandidate.get, requestModel, indexAliasOption, factGroupByKeys, queryAttributes)
case DimFactQuery =>
require(factBestCandidate.isDefined, "dim fact query should have fact defined")
CombinedQueryContext(dims, factBestCandidate.get, requestModel, queryAttributes)
Expand Down
32 changes: 21 additions & 11 deletions core/src/main/scala/com/yahoo/maha/core/query/QueryPipeline.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,16 @@ object QueryPipeline extends Logging {
val asyncDisqualifyingSet: Set[Engine] = Set(DruidEngine)

val completeRowList: Query => RowList = (q) => new CompleteRowList(q)
val dimDrivenPartialRowList: Query => RowList = (q) => new DimDrivenPartialRowList(q.queryContext.indexAliasOption.get, q)
val dimDrivenFactOrderedPartialRowList: Query => RowList = (q) => new DimDrivenFactOrderedPartialRowList(q.queryContext.indexAliasOption.get, q)
val factDrivenPartialRowList: Query => RowList = (q) => new FactDrivenPartialRowList(q.queryContext.indexAliasOption.get, q)
val dimDrivenPartialRowList: Query => RowList = (q) => {
new DimDrivenPartialRowList(RowGrouping(q.queryContext.indexAliasOption.get, List(q.queryContext.indexAliasOption.get)), q)
}
val dimDrivenFactOrderedPartialRowList: Query => RowList = (q) => {
new DimDrivenFactOrderedPartialRowList(RowGrouping(q.queryContext.indexAliasOption.get, List(q.queryContext.indexAliasOption.get)), q)}
val factDrivenPartialRowList: Query => RowList = (q) => {
val indexAlias = q.queryContext.indexAliasOption.get
val groupByKeys: List[String] = q.queryContext.factGroupByKeys
new FactDrivenPartialRowList(RowGrouping(indexAlias, groupByKeys), q)
}

def unionViewPartialRowList(queryList: List[Query]): Query => RowList = {
require(queryList.forall(q => q.queryContext.isInstanceOf[FactualQueryContext] && q.queryContext.asInstanceOf[FactualQueryContext].factBestCandidate.fact.isInstanceOf[ViewBaseTable])
Expand Down Expand Up @@ -86,7 +93,7 @@ object QueryPipeline extends Logging {
case any =>
throw new UnsupportedOperationException(s"Cannot create UnionViewRowList with fact of type : ${any.getClass.getSimpleName}")
}
case DimQueryContext(_, _, _, _) =>
case DimQueryContext(_, _, _, _, _) =>
throw new IllegalArgumentException(s"Requested UnionViewRowList in DimQueryContext")
case _ =>
throw new IllegalArgumentException(s"Requested UnionViewRowList in Unhandled/UnknownQueryContext")
Expand Down Expand Up @@ -675,24 +682,27 @@ class DefaultQueryPipelineFactory(implicit val queryGeneratorRegistry: QueryGene
private[this] def getMultiEngineDimQuery(bestDimCandidates: SortedSet[DimensionBundle],
requestModel: RequestModel,
indexAlias: String,
factGroupByKeys: List[String],
queryAttributes: QueryAttributes,
queryGenVersion: Version): Query = {
val dimOnlyContextBuilder = QueryContext
.newQueryContext(DimOnlyQuery, requestModel)
.addDimTable(bestDimCandidates)
.addIndexAlias(indexAlias)
.addFactGroupByKeys(factGroupByKeys)
.setQueryAttributes(queryAttributes)

require(queryGeneratorRegistry.isEngineRegistered(bestDimCandidates.head.dim.engine, Option(queryGenVersion))
, s"Failed to find query generator for engine : ${bestDimCandidates.head.dim.engine}")
queryGeneratorRegistry.getValidGeneratorForVersion(bestDimCandidates.head.dim.engine, queryGenVersion, Option(requestModel)).get.generate(dimOnlyContextBuilder.build())
}

private[this] def getFactQuery(bestFactCandidate: FactBestCandidate, requestModel: => RequestModel, indexAlias: String, queryGenVersion: Version): Query = {
private[this] def getFactQuery(bestFactCandidate: FactBestCandidate, requestModel: => RequestModel, indexAlias: String, factGroupByKeys: List[String], queryGenVersion: Version): Query = {
val factOnlyContextBuilder = QueryContext
.newQueryContext(FactOnlyQuery, requestModel)
.addFactBestCandidate(bestFactCandidate)
.addIndexAlias(indexAlias)
.addFactGroupByKeys(factGroupByKeys)
require(queryGeneratorRegistry.isEngineRegistered(bestFactCandidate.fact.engine, Some(queryGenVersion))
, s"Failed to find query generator for engine : ${bestFactCandidate.fact.engine}")
queryGeneratorRegistry.getValidGeneratorForVersion(bestFactCandidate.fact.engine, queryGenVersion, Option(requestModel)).get.generate(factOnlyContextBuilder.build())
Expand Down Expand Up @@ -851,14 +861,14 @@ OuterGroupBy operation has to be applied only in the following cases
if (!requestModel.hasFactSortBy && requestModel.forceDimDriven) {
//oracle + druid
requestDebug("dimQueryThenFactQuery")
val dimQuery = getMultiEngineDimQuery(bestDimCandidates, requestModel, indexAlias, queryAttributes, queryGenVersion)
val dimQuery = getMultiEngineDimQuery(bestDimCandidates, requestModel, indexAlias, List(indexAlias), queryAttributes, queryGenVersion)
val subsequentQuery: (IndexedRowList, QueryAttributes) => Query = {
case (irl, subqueryAttributes) =>
val field = irl.indexAlias
val field = irl.rowGrouping.indexAlias
val values = irl.keys.toList.map(_.toString)
val filter = InFilter(field, values)
val injectedFactBestCandidate = factOnlyInjectFilter(factBestCandidateOption.get, filter)
val query = getFactQuery(injectedFactBestCandidate, requestModel, indexAlias, queryGenVersion)
val query = getFactQuery(injectedFactBestCandidate, requestModel, indexAlias, List(indexAlias), queryGenVersion)
irl.addSubQuery(query)
query
}
Expand All @@ -879,10 +889,10 @@ OuterGroupBy operation has to be applied only in the following cases
//since druid + oracle doesn't support row count
requestDebug("factQueryThenDimQuery")
val noRowCountRequestModel = requestModel.copy(includeRowCount = false)
val factQuery = getFactQuery(factBestCandidateOption.get, noRowCountRequestModel, indexAlias, queryGenVersion)
val factQuery = getFactQuery(factBestCandidateOption.get, noRowCountRequestModel, indexAlias, List(indexAlias), queryGenVersion)
val subsequentQuery: (IndexedRowList, QueryAttributes) => Query = {
case (irl, subqueryqueryAttributes) =>
val field = irl.indexAlias
val field = irl.rowGrouping.indexAlias
val values = irl.keys.toList.map(_.toString)
val valuesSize = values.size
if (values.nonEmpty && (valuesSize >= noRowCountRequestModel.maxRows || noRowCountRequestModel.startIndex <= 0)) {
Expand All @@ -896,7 +906,7 @@ OuterGroupBy operation has to be applied only in the following cases
}
queryAttributesBuilder.build
}
getMultiEngineDimQuery(bestDimCandidates, noRowCountRequestModel, indexAlias, injectedAttributes, queryGenVersion)
getMultiEngineDimQuery(bestDimCandidates, noRowCountRequestModel, indexAlias, List(indexAlias), injectedAttributes, queryGenVersion)
} else {
QueryChain.logger.info("No data returned from druid, should run fallback query if there is one")
NoopQuery
Expand Down
Loading