Reduce/Fold only some elements

Question

I have a file parser that produces a collection of elements all belonging to the same trait. It is similar to the following.

trait Data {
  val identifier: String
}

case class Meta(identifier: String, props: Properties) extends Data
case class Complete(identifier: String, contents: Map[String, Any]) extends Data
case class Partial(identifier: String, name: String, value: Any) extends Data

... 

def parse(file: File): Iterator[Data] = ... // this isn't relevant

What I am attempting to do is traverse the collection in a functional manner since I am processing a lot of data and want to be as memory conscious as possible. The collection when it is returned from the parse method is a mix of Complete, Meta, and Partial elements. The logic is that I need to pass the Complete and Meta elements through unchanged, while collecting the Partial elements and grouping on the identifier to create Complete elements.

With just a collection of Partial elements (Iterator[Partial]), I can do the following:

partialsOnly.groupBy(_.identifier)
 .map{ 
   case (ident, parts) => 
     Complete(ident, parts.map(p => p.name -> p.value).toMap)
 }

Is there a functional way, somewhat similar to scan that will accumulate elements, but only some elements, while letting the rest through unchanged?

user_4268347 · Answer

You can use the partition function to split a collection in two based on a predicate.

val (partial: List[Data], completeAndMeta: List[Data]) = parse("file").partition(_ match{
  case partial: Partial => true
  case _ => false
})

From there, you want to make sure you can process partial as a List[Partial], ideally without tripping compiler warnings about type erasure or doing messy casts. You can do this with a call to collect, using a function that only accepts Partial's.

val partials: List[Partial] = partial.collect(_.match{case partial: Partial => partial}}

Unfortunately, when used on an Iterator, partition may need to buffer arbitrary amounts of data, so isn't necessarily the most memory efficient technique. If memory management is a huge concern, you may need to sacrifice functional purity. Alternately, if you add some way of knowing when a Partial is completed, you can accumulate them in a Map via a foldLeft and emit the final value as they finish.

user_1303324 · Answer

Recursion might be functional way to solve your problem:

def parse(list: List[Data]): (List[Data], List[Data]) = {
  list match {
     case (x:Partial) :: xs =>
       val (partials, rest) = parse(xs)
       (x :: partials, rest) //instead of creating list, you can join partials here
     case x :: xs =>
       val (partials, rest) = parse(xs)
       (partials, x :: rest)
     case _ => (Nil, Nil)
   }
}

val (partials, rest) = parse(list)

Unfortunately, this function is not tail recursive, so it might blow up the stack for longer lists.

You can solve it by using Eval from cats:

def parse2(list: List[Data]): Eval[(List[Data], List[Data])] =
    Eval.now(list).flatMap {
      case (x:Partial) :: xs =>
        parse2(xs).map {
          case (partials, rest) => (x :: partials, rest) //instead of creating list, you can join partials here
        }
      case x :: xs =>
        parse2(xs).map {
          case (partials, rest) => (partials, x :: rest)
        }
      case _ => Eval.now((Nil, Nil))
    }

val (partialsResult, restResult) = parse2(longList).value

This solution would be safe for the stack because it uses Heap, not Stack.

And here's version, which also groups partials:

def parse3(list: List[Data]): Eval[(Map[String, List[Partial]], List[Data])] =
Eval.now(list).flatMap {
  case (x:Partial) :: xs =>
    parse3(xs).map {
      case (partials, rest) =>
        val newPartials = x :: partials.getOrElse(x.identifier, Nil)
        (partials + (x.identifier -> newPartials), rest)
    }
  case x :: xs =>
    parse3(xs).map {
      case (partials, rest) => (partials, x :: rest)
    }
  case _ => Eval.now((Map.empty[String, List[Partial]], Nil))
}

Reduce/Fold only some elements

2 Answers