Adding RangeVectorTransformer for le filter for PeriodicSeriesWithWin…

…dowing
filodb · Feb 14, 2024 · d4cd73b · d4cd73b
1 parent 32e4eeb
commit d4cd73b
Showing 1 changed file with 16 additions and 4 deletions.
diff --git a/coordinator/src/main/scala/filodb.coordinator/queryplanner/SingleClusterPlanner.scala b/coordinator/src/main/scala/filodb.coordinator/queryplanner/SingleClusterPlanner.scala
@@ -558,13 +558,16 @@ class SingleClusterPlanner(val dataset: Dataset,
     }
   }
 
+  // scalastyle:off method.length
   override private[queryplanner] def materializePeriodicSeriesWithWindowing(qContext: QueryContext,
                                                      lp: PeriodicSeriesWithWindowing,
                                                      forceInProcess: Boolean): PlanResult = {
 
-    val logicalPlanWithoutBucket = if (queryConfig.translatePromToFilodbHistogram) {
-       removeBucket(Right(lp))._3.right.get
-    } else lp
+    val (nameFilter: Option[String], leFilter: Option[String], logicalPlanWithoutBucket: PeriodicSeriesWithWindowing) =
+      if (queryConfig.translatePromToFilodbHistogram) {
+       val result = removeBucket(Right(lp))
+        (result._1, result._2, result._3.right.get)
+    } else (None, None, lp)
 
     val series = walkLogicalPlanTree(logicalPlanWithoutBucket.series, qContext, forceInProcess)
     val rawSource = logicalPlanWithoutBucket.series.isRaw
@@ -585,6 +588,15 @@ class SingleClusterPlanner(val dataset: Dataset,
       realScanStepMs, realScanEndMs, window, Some(execRangeFn), qContext,
       logicalPlanWithoutBucket.stepMultipleNotationUsed,
       paramsExec, logicalPlanWithoutBucket.offsetMs, rawSource)))
+
+    // Add the le filter transformer to select the required bucket
+    if (nameFilter.isDefined && nameFilter.head.endsWith("_bucket") && leFilter.isDefined) {
+      val paramsExec = StaticFuncArgs(leFilter.head.toDouble, RangeParams(realScanStartMs / 1000, realScanStepMs / 1000,
+        realScanEndMs / 1000))
+      series.plans.foreach(_.addRangeVectorTransformer(InstantVectorFunctionMapper(HistogramBucket,
+        Seq(paramsExec))))
+    }
+
     val result = if (logicalPlanWithoutBucket.function == RangeFunctionId.AbsentOverTime) {
       val aggregate = Aggregate(AggregationOperator.Sum, logicalPlanWithoutBucket, Nil,
                                 AggregateClause.byOpt(Seq("job")))
@@ -602,9 +614,9 @@ class SingleClusterPlanner(val dataset: Dataset,
       result.plans.foreach(p => p.addRangeVectorTransformer(RepeatTransformer(lp.startMs, lp.stepMs, lp.endMs,
         p.queryWithPlanName(qContext))))
     }
-
     result
   }
+  // scalastyle:on method.length
 
   override private[queryplanner] def removeBucket(lp: Either[PeriodicSeries, PeriodicSeriesWithWindowing]) = {
     val rawSeries = lp match {