//process input row and produce action row to insert into NORMAQM
  def getIdsThenInsertAction: FDAUserTask[FDAROW] = row => { row match { case aqm: AQMRPTModel =>
        if (aqm.valid) { val stateId = getStateID(aqm.state) val countyId = getCountyID(aqm.state,aqm.county) val action = NORMAQMQuery += NORMAQMModel(0,aqm.mid, stateId, countyId, aqm.year,aqm.value,aqm.total) fda_next(FDAActionRow(action)) } else fda_skip case _ => fda_skip } } //runner for the action rows
  val runner = FDAActionRunner(slick.jdbc.H2Profile) def runInsertAction: FDAUserTask[FDAROW] = row => row match { case FDAActionRow(action) => runner.fda_execAction(action)(db) fda_skip case _ => fda_skip }

 //create parallel sources //get a stream of years
  val qryYears = AQMRPTQuery.map(_.year).distinct case class Years(year: Int) extends FDAROW implicit def toYears(y: Int) = Years(y) val yearViewLoader = FDAViewLoader(slick.jdbc.H2Profile)(toYears _) val yearSeq = yearViewLoader.fda_typedRows(qryYears.result)(db).toSeq val yearStream = fda_staticSource(yearSeq)()

  //strong row type
  implicit def toAQMRPT(row: AQMRPTTable#TableElementType) = AQMRPTModel(row.rid, row.mid, row.state, row.county, row.year, row.value, row.total, row.valid) //shared stream loader when operate in parallel mode
  val AQMRPTLoader = FDAStreamLoader(slick.jdbc.H2Profile)(toAQMRPT _) //loading rows with year yr
  def loadRowsInYear(yr: Int) = { //a new query
    val query = AQMRPTQuery.filter(row => row.year === yr) //reuse same loader
    AQMRPTLoader.fda_typedStream(query.result)(db)(256, 256)() }

  //loading rows by year
  def loadRowsByYear: FDASourceLoader = row => { row match { case Years(y) => loadRowsInYear(y) //produce stream of the year
      case _ => fda_appendRow(FDANullRow) } }

  //get parallel source constructor
  val parSource = yearStream.toParSource(loadRowsByYear)

  //produce a stream from parallel sources
  val source = fda_par_source(parSource)(3)

  //the following is a process of composition of stream combinators //get parallel source constructor
  val parSource = yearStream.toParSource(loadRowsByYear) //implicit val strategy = Strategy.fromCachedDaemonPool("cachedPool") //produce a stream from parallel sources
  val source = fda_par_source(parSource)(3) //turn getIdsThenInsertAction into parallel task
  val parTasks = source.toPar(getIdsThenInsertAction) //runPar to produce a new stream
  val actionStream =fda_runPar(parTasks)(3) //turn runInsertAction into parallel task
  val parRun = actionStream.toPar(runInsertAction) //runPar and carry out by startRun
  fda_runPar(parRun)(2).startRun