Skip to content
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

hack for GRPC execution plans #1684

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

hack for GRPC execution plans #1684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
16 changes: 11 additions & 5 deletions coordinator/src/main/scala/filodb.coordinator/ActorPlanDispatcher.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,14 @@
package filodb.coordinator

import java.util.concurrent.TimeUnit

import scala.concurrent.duration.FiniteDuration

import akka.actor.ActorRef
import akka.pattern.{ask, AskTimeoutException}
import akka.pattern.{AskTimeoutException, ask}
import akka.util.Timeout
import filodb.coordinator.client.QueryCommands.ProtoExecPlan
import monix.eval.Task
import monix.execution.Scheduler
import monix.reactive.Observable

import filodb.core.QueryTimeoutException
import filodb.core.query.{QueryStats, QueryWarnings, ResultSchema}
import filodb.core.store.ChunkSource
Expand Down Expand Up @@ -44,7 +42,15 @@ case class ActorPlanDispatcher(target: ActorRef, clusterName: String) extends Pl
emptyPartialResult
})
} else {
val fut = (target ? plan.execPlan) (t).map {
// HACK
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this really a hack?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, this is a hack, not meant to be committed, this PR is just a POC that shows how proto exec plans can works end to end and solicit the feedback. Here, for example the question is how would we want to proceed with the configuration switch that would enable proto serialization for the exec plans. I think the most preferable option is to pass it as a query parameter, smth like protoExec=true.

val doProto = plan.execPlan.queryContext.plannerParams.warnLimits.groupByCardinality == 7777
val message = if (doProto) {
val protoPlan = ProtoConverters.execPlanToProto(plan.execPlan)
ProtoExecPlan(plan.execPlan.dataset, protoPlan.toByteArray, plan.execPlan.submitTime)
} else {
plan.execPlan
}
val fut = (target ? message) (t).map {
case resp: QueryResponse => resp
case e => throw new IllegalStateException(s"Received bad response $e")
}
Expand Down
13 changes: 10 additions & 3 deletions coordinator/src/main/scala/filodb.coordinator/QueryActor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import scala.collection.mutable
import scala.concurrent.duration.{DurationInt, FiniteDuration}
import scala.util.{Failure, Success}
import scala.util.control.NonFatal

import akka.actor.{ActorRef, Props}
import akka.pattern.AskTimeoutException
import kamon.Kamon
Expand All @@ -14,7 +13,6 @@ import monix.eval.Task
import monix.execution.exceptions.ExecutionRejectedException
import net.ceedubs.ficus.Ficus._
import net.ceedubs.ficus.readers.ValueReader

import filodb.coordinator.queryplanner.SingleClusterPlanner
import filodb.core._
import filodb.core.memstore.{FiloSchedulers, TermInfo, TimeSeriesStore}
Expand All @@ -27,6 +25,7 @@ import filodb.core.query.QuerySession
import filodb.core.query.QueryStats
import filodb.core.query.SerializedRangeVector
import filodb.core.store.CorruptVectorException
import filodb.grpc.ExecPlans.ExecPlanContainer
import filodb.query._
import filodb.query.exec.{ExecPlan, InProcessPlanDispatcher, PlanDispatcher}

Expand Down Expand Up @@ -293,12 +292,20 @@ final class QueryActor(memStore: TimeSeriesStore,
}
}

def execProtoExecPlan(pep: ProtoExecPlan, replyTo: ActorRef): Unit = {
import filodb.coordinator.ProtoConverters._
val c = ExecPlanContainer.parseFrom(pep.serializedExecPlan)
val plan: ExecPlan = c.fromProto()
execPhysicalPlan2(plan, replyTo)
}

def receive: Receive = {
case q: LogicalPlan2Query => val replyTo = sender()
processLogicalPlan2Query(q, replyTo)
case q: ExplainPlan2Query => val replyTo = sender()
processExplainPlanQuery(q, replyTo)
case q: ExecPlan => execPhysicalPlan2(q, sender())
case q: ExecPlan => execPhysicalPlan2(q, sender())
case q: ProtoExecPlan => execProtoExecPlan(q, sender())
case q: GetTopkCardinality => execTopkCardinalityQuery(q, sender())

case GetIndexNames(ref, limit, _) =>
Expand Down
4 changes: 4 additions & 0 deletions coordinator/src/main/scala/filodb.coordinator/client/QueryCommands.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,10 @@ object QueryCommands {
logicalPlan: LogicalPlan2,
qContext: QueryContext = QueryContext(),
submitTime: Long = System.currentTimeMillis()) extends QueryCommand

final case class ProtoExecPlan(dataset: DatasetRef,
serializedExecPlan: Array[Byte],
submitTime: Long = System.currentTimeMillis()) extends QueryCommand
// Error responses from query
final case class UndefinedColumns(undefined: Set[String]) extends ErrorResponse
final case class BadArgument(msg: String) extends ErrorResponse with QueryResponse
Expand Down
Loading
Loading