Slick Query in stream

/*
 * Assume we have a table (TableQuery) called `bigDataTable` correctly defined including a blob type.
 * Under postgres, it's implemented using the lo extension, and
 * interactions with this table are only supported using transactions.
 *
 * Also assume we have a table called `modelDataTable` which has a blob type as well, so inserts are only done using
 *
 */

/**
 * returns a stream of rows from the table.  Table contains way more data
 * than can be in memory.  DataAccess defines a database method which returns the database connection we're using,
 * and makes the right profile available.
 */
trait BigDataTableQL { this : DataAccess =>
    import profile.simple._
    import profile.Implicits

    def getSomeDataQ = for{
        item <- bigDataTable
    } yield item

    lazy val getSomeData = Compiled(getSomeDataQ.to[Stream])
}

/**
 * Now we're going to use this stream.
 */

object Main extends App with BigDataTableQL with ProductionDataAccess{

    def isInteresting(b : BigDataTableItem) : Boolean  = {
        //return true sometimes.
    }
    /*
     *  This will consume the entire stream, and build a giant bunch of inserts, and not hit the
     *  database until the entire stream is consumed, which will result in out of memory error.
     */
    def analyzeAttempt : Unit = {
        database withTransaction { implicit s =>
            getSomeData.run.grouped(1000).foreach{ toInsert =>
                val thisInsertGroup = toInsert.filter(isInteresting(_)).map(compileModelData(_))
                modelDataTable ++= thisInsertGroup.toList
            }
        } //Transaction never completes, out of memory error as entire stream is consumed...
    }

    /**
     * For some reason, nesting another transaction here doesn't work either.
     *
     */
    def analyzeAttempt2 : Unit = {
        database withTransaction{ implicit s =>
            getSomeData.run.grouped(1000).foreach{ toInsert =>
                database withTransaction {implicit s2 =>
                    val thisInsertGroup = toInsert.filter(isInteresting(_)).map(compileModelData(_))
                    modelDataTable.++=(thisInsertGroup.toList)(s2)
                }   //WANT A DATABASE HIT HERE TO INSERT ABOUT 1000 ROWS AT A TIME.  Doesn't happen.
            }
        }
    }
}