message: abstract next_loc
diff: @@ -154,21 +154,14 @@ impl<'a> SqliteSourcePartitionParser<'a> {
  OwningHandle::new_with_fn(Box::new(stmt), |stmt: *const Statement<'a>| unsafe {
  DummyBox((&mut *(stmt as *mut Statement<'_>)).query([]).unwrap())
  });
  -
  Ok(Self {
  rows,
  ncols: schema.len(),
  current_col: 0,
  })
  }
  -}
  -impl<'a> PartitionParser<'a> for SqliteSourcePartitionParser<'a> {
  - type TypeSystem = SqliteTypeSystem;
  -}
  -
  -impl<'r, 'a> Produce<'r, i32> for SqliteSourcePartitionParser<'a> {
  - fn produce(&'r mut self) -> Result<i32> {
  + fn next_loc(&mut self) -> Result<(&Row, usize)> {
  let row: &Row = match self.current_col {
  0 => self.rows.next()?.ok_or_else(|| anyhow!("Sqlite EOF"))?,
  _ => self
  @@ -176,9 +169,20 @@ impl<'r, 'a> Produce<'r, i32> for SqliteSourcePartitionParser<'a> {
  .get()
  .ok_or_else(|| anyhow!("Sqlite empty current row"))?,
  };
  -
  - let val = row.get(self.current_col)?;
  + let col = self.current_col;
  self.current_col = (self.current_col + 1) % self.ncols;
  + Ok((row, col))
  + }
  +}
  +
  +impl<'a> PartitionParser<'a> for SqliteSourcePartitionParser<'a> {
  + type TypeSystem = SqliteTypeSystem;
  +}
  +
  +impl<'r, 'a> Produce<'r, i32> for SqliteSourcePartitionParser<'a> {
  + fn produce(&'r mut self) -> Result<i32> {
  + let (row, col) = self.next_loc()?;
  + let val = row.get(col)?;
  Ok(val)
  }
  }
  