feat(misc): Adding new helper method to get column filter list along with leaf plan

coordinator/src/test/scala/filodb.coordinator/queryplanner/LogicalPlanUtilsSpec.scala

@@ -107,16 +107,16 @@ class LogicalPlanUtilsSpec   extends AnyFunSpec with Matchers {
         Seq("A", "C"), Seq("A", "D"), Seq("B", "C"), Seq("B", "D")))
     )
     val shouldErrorQueries = Seq(
-      """foo{label1=~"A|B", label2=~"C|D."}""",  // non-pipe regex chars
-      """foo{label1=~"A|B", label2!~"C|D"}"""    // non-equals[regex] filter
+      """foo{label1=~"A|B", label2=~"C|D."}""", // non-pipe regex chars
+      """foo{label1=~"A|B", label2!~"C|D"}""" // non-equals[regex] filter
     )
     for ((query, expected) <- queryExpectedPairs) {
       val plan = Parser.queryToLogicalPlan(query, 100, 10)
       val res = LogicalPlanUtils.resolvePipeConcatenatedShardKeyFilters(plan, shardKeyLabels)
       // make sure all filters are Equals
       res.foreach(_.foreach(_.filter.isInstanceOf[Equals] shouldEqual true))
       // make sure all values are as expected
-      res.map{ group =>
+      res.map { group =>
         group.sortBy(_.column).map(_.filter.valuesStrings.head)
       }.toSet shouldEqual expected
     }
@@ -128,4 +128,31 @@ class LogicalPlanUtilsSpec   extends AnyFunSpec with Matchers {
       }
     }
   }
+
+  it ("getColumnFilterGroupWithLogical plans should return result as expected") {
+    val timeParamsSec = TimeStepParams(1000, 10, 10000)
+    val query1 = """sum(count_over_time((test_metric{_ws_="test-ws", _ns_="test-ns", usecase="test"} > 20000)[21600s:])) or vector(0)"""
+    val query2 = """sum(count_over_time((test_metric{_ws_="test-ws", _ns_="test-ns", usecase="test"} > 20000)[21600s:])) or vector(0) or sum(count_over_time((test_metric{_ws_="wrong_ws", _ns_="wrong-ns", usecase="test"} > 20000)[21600s:]))"""
+    val query3 = """rate(test_metric{_ns_="test-ns", _ws_="test-ns", cluster="test1"}[5m]) * 1000"""
+
+    def getColumnFilterWithAndWithoutScalarCount(query: String) : (Int, Int) = {
+      val lp = Parser.queryRangeToLogicalPlan(query, timeParamsSec)
+      val columnGroups = LogicalPlan.getColumnFilterGroup(lp)
+      val columnGroupsWithoutScalar = LogicalPlan.getColumnFilterGroupFilteringLeafScalarPlans(lp)
+      val scalarPlansCount = columnGroups.count(x => x.isEmpty)
+      val scalarPlansWithoutCount = columnGroupsWithoutScalar.count(x => x.isEmpty)
+      (scalarPlansCount, scalarPlansWithoutCount)
+    }
+    var counts = getColumnFilterWithAndWithoutScalarCount(query1)
+    counts._1 shouldEqual 2
+    counts._2 shouldEqual 0
+
+    counts = getColumnFilterWithAndWithoutScalarCount(query2)
+    counts._1 shouldEqual 3
+    counts._2 shouldEqual 0
+
+    counts = getColumnFilterWithAndWithoutScalarCount(query3)
+    counts._1 shouldEqual 1
+    counts._2 shouldEqual 0
+  }
 }

query/src/main/scala/filodb/query/LogicalPlan.scala

@@ -761,6 +761,40 @@ object LogicalPlan {
     }
   }
 
+  /**
+   * Given a LogicalPlan, the function finds a Seq of all Child nodes,and returns a Set of ColumnFilters.
+   * It filters out the column filter group for all the scalar leaf plans
+   *
+   * @param logicalPlan the root LogicalPlan
+   * @return Seq[ColumnFilterListWithLogicalPlan], Seq has size same as the number of leaf nodes
+   */
+  def getColumnFilterGroupFilteringLeafScalarPlans(logicalPlan: LogicalPlan): Seq[Set[ColumnFilter]] = {
+    LogicalPlan.findLeafLogicalPlans(logicalPlan) map { lp =>
+      lp match {
+        case lp: LabelValues => (lp.filters toSet, true)
+        case lp: LabelNames => (lp.filters toSet, true)
+        case lp: RawSeries => (lp.filters toSet, true)
+        case lp: RawChunkMeta => (lp.filters toSet, true)
+        case lp: SeriesKeysByFilters => (lp.filters toSet, true)
+        case lp: LabelCardinality => (lp.filters.toSet, true)
+        case lp: TsCardinalities => (lp.filters.toSet, true)
+        case _: ScalarTimeBasedPlan => (Set.empty[ColumnFilter], false)
+        case _: ScalarFixedDoublePlan => (Set.empty[ColumnFilter], false)
+        case _: ScalarBinaryOperation => (Set.empty[ColumnFilter], false)
+        case _ => throw new BadQueryException(s"Invalid logical plan $logicalPlan")
+      }
+    } match {
+      case groupSeq: Seq[(Set[ColumnFilter], Boolean)] =>
+        if (groupSeq.isEmpty || groupSeq.forall( x => x._1.isEmpty)) Seq.empty else {
+          // filter out the scalar plans with empty column filter group
+          // and only include the column filter of raw leaf plans
+          groupSeq.filter(y => y._2)
+            .map(x => x._1)
+        }
+      case _ => Seq.empty
+    }
+  }
+
   def getRawSeriesFilters(logicalPlan: LogicalPlan): Seq[Seq[ColumnFilter]] = {
     LogicalPlan.findLeafLogicalPlans(logicalPlan).map { l =>
       l match {