diff --git a/csv/shared/src/main/scala/fs2/data/csv/CsvRow.scala b/csv/shared/src/main/scala/fs2/data/csv/CsvRow.scala
deleted file mode 100644
index 75180e6b2..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/CsvRow.scala
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import cats.data.NonEmptyList
-import cats.syntax.all._
-
-object CsvRow {
-
-  /** Constructs a [[CsvRow]] and checks that the size of values and headers match. */
-  def apply[Header](values: NonEmptyList[String],
-                    headers: NonEmptyList[Header],
-                    line: Option[Long] = None): Either[CsvException, CsvRow[Header]] =
-    if (values.length =!= headers.length)
-      Left(new HeaderSizeError(headers.length, values.length, line))
-    else
-      Right(new CsvRow(values, Some(headers), line))
-
-  def unsafe[Header](values: NonEmptyList[String], headers: NonEmptyList[Header]): CsvRow[Header] =
-    apply(values, headers).fold(throw _, identity)
-
-  def liftRow[Header](headers: NonEmptyList[Header])(row: Row): Either[CsvException, CsvRow[Header]] =
-    apply(row.values, headers, row.line)
-
-  def fromListHeaders[Header](l: List[(Header, String)]): Option[CsvRow[Header]] = {
-    val (hs, vs) = l.unzip
-    (NonEmptyList.fromList(vs), NonEmptyList.fromList(hs)).mapN((v, h) => new CsvRow(v, Some(h)))
-  }
-
-  def fromNelHeaders[Header](nel: NonEmptyList[(Header, String)]): CsvRow[Header] = {
-    val (hs, vs) = nel.toList.unzip
-    new CsvRow(NonEmptyList.fromListUnsafe(vs), Some(NonEmptyList.fromListUnsafe(hs)))
-  }
-
-  def unapply[Header](arg: CsvRow[Header]): Some[(NonEmptyList[String], NonEmptyList[Header])] = Some(
-    (arg.values, arg.headers.get))
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/CsvRowDecoder.scala b/csv/shared/src/main/scala/fs2/data/csv/CsvRowDecoder.scala
index 964a206c7..ca6cce3eb 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/CsvRowDecoder.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/CsvRowDecoder.scala
@@ -1,27 +1,53 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
 package fs2.data.csv
 
+import cats.syntax.all._
+import cats._
+
+sealed trait CsvRowDecoder[T] {
+  def apply(headers: List[String]): Either[DecoderError, RowDecoder[T]]
+}
+
 object CsvRowDecoder {
+  def as[T: CellDecoder](name: String): CsvRowDecoder[T] = new CsvRowDecoder[T] {
+    def apply(headers: List[String]) = {
+      val i = headers.toList.indexOf(name)
+      Either.cond(
+        i >= 0,
+        RowDecoder.instance { (row: Row) =>
+          row.values.toList.lift(i).toRight {
+            new HeaderSizeError(headers.size, row.values.size, row.line)
+          }.flatMap(CellDecoder[T].apply)
+        },
+        new DecoderError(s"unknown field $name")
+      )
+    }
+  }
+
+  private[csv] def instance[T](f: List[String] => Either[DecoderError, RowDecoder[T]]): CsvRowDecoder[T] = new CsvRowDecoder[T] {
+    def apply(headers: List[String]) = f(headers)
+  }
 
-  @inline
-  def apply[T: CsvRowDecoder[*, Header], Header]: CsvRowDecoder[T, Header] = implicitly[CsvRowDecoder[T, Header]]
+  implicit def decodeResultCsvRowDecoder[T](implicit
+      dec: CsvRowDecoder[T]): CsvRowDecoder[DecoderResult[T]] =
+    new CsvRowDecoder[DecoderResult[T]] {
+      override def apply(headers: List[String]): Either[DecoderError,RowDecoder[DecoderResult[T]]] =
+        dec(headers).map(_.map(_.asRight))
+    }
 
-  @inline
-  def instance[T, Header](f: CsvRow[Header] => DecoderResult[T]): CsvRowDecoder[T, Header] = f(_)
+  implicit val CsvRowDecoderInstances: Applicative[CsvRowDecoder] with SemigroupK[CsvRowDecoder] =
+    new Applicative[CsvRowDecoder] with SemigroupK[CsvRowDecoder] {
+      val a = Applicative[List[String] => *].compose(Applicative[Either[DecoderError, *]].compose(Applicative[RowDecoder]))
+      override def combineK[A](x: CsvRowDecoder[A], y: CsvRowDecoder[A]): CsvRowDecoder[A] =
+        instance((x.apply, y.apply).mapN(_ <+> _))
+        
+      override def map[A, B](fa: CsvRowDecoder[A])(f: A => B): CsvRowDecoder[B] =
+        instance(a.map(fa.apply)(f))
 
+      def pure[A](x: A): CsvRowDecoder[A] =
+        instance(a.pure(x))
+      
+      def ap[A, B](ff: CsvRowDecoder[A => B])(fa: CsvRowDecoder[A]): CsvRowDecoder[B] = {
+        instance(a.ap(ff.apply)(fa.apply))
+      }
+    }
 }
diff --git a/csv/shared/src/main/scala/fs2/data/csv/CsvRowEncoder.scala b/csv/shared/src/main/scala/fs2/data/csv/CsvRowEncoder.scala
index a183546e6..f3f8ed84e 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/CsvRowEncoder.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/CsvRowEncoder.scala
@@ -1,29 +1,53 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
 package fs2.data.csv
 
-import cats.data.NonEmptyList
+import cats.ContravariantMonoidal
+import cats.syntax.all._
+
+sealed trait CsvRowEncoder[T] {
+  def headers: List[String]
+  def encoder: RowEncoder[T]
+}
 
 object CsvRowEncoder {
+  private[csv] def instance[T](
+    headers: List[String],
+    encoder: RowEncoder[T]
+  ) = {
+    def h = headers
+    def e = encoder
+    new CsvRowEncoder[T] {
+      def headers = h
+      def encoder = e
+    }
+  }
+
+  def apply[T](implicit ev: CsvRowEncoder[T]): CsvRowEncoder[T] = ev
+
+  implicit val csvRowEncoderInstances = new ContravariantMonoidal[CsvRowEncoder] {
+
+    override def product[A, B](fa: CsvRowEncoder[A], fb: CsvRowEncoder[B]): CsvRowEncoder[(A, B)] =
+      new CsvRowEncoder[(A, B)] {
+        override def headers: List[String] =
+          fa.headers |+| fb.headers.toList
+
+        override def encoder: RowEncoder[(A, B)] =
+          (fa.encoder, fb.encoder).tupled
+
+      }
+
+    override def contramap[A, B](fa: CsvRowEncoder[A])(f: B => A): CsvRowEncoder[B] =
+      new CsvRowEncoder[B] {
+
+        override def headers: List[String] = fa.headers
+
+        override def encoder: RowEncoder[B] = fa.encoder.contramap(f)
 
-  @inline
-  def apply[T: CsvRowEncoder[*, Header], Header]: CsvRowEncoder[T, Header] = implicitly[CsvRowEncoder[T, Header]]
+      }
 
-  @inline
-  def instance[T, Header](f: T => NonEmptyList[(Header, String)]): CsvRowEncoder[T, Header] =
-    (t: T) => CsvRow.fromNelHeaders(f(t))
+    override def unit: CsvRowEncoder[Unit] =
+      new CsvRowEncoder[Unit] {
+        def headers = Nil
+        def encoder = ContravariantMonoidal[RowEncoder].unit
+      }
+  }
 }
diff --git a/csv/shared/src/main/scala/fs2/data/csv/HasHeaders.scala b/csv/shared/src/main/scala/fs2/data/csv/HasHeaders.scala
deleted file mode 100644
index 898db7b9e..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/HasHeaders.scala
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-/** Witness that a [[RowF]] has headers of a certain type.
-  * @tparam H
-  * @tparam Header
-  */
-sealed trait HasHeaders[H[+a] <: Option[a], Header] extends (RowF[H, Header] => CsvRow[Header])
-
-object HasHeaders {
-  implicit def hasHeaders[Header]: HasHeaders[Some, Header] = new HasHeaders[Some, Header] {
-    override def apply(value: RowF[Some, Header]): CsvRow[Header] = value
-  }
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/Row.scala b/csv/shared/src/main/scala/fs2/data/csv/Row.scala
index 453b156cc..29801c51c 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/Row.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/Row.scala
@@ -16,9 +16,92 @@
 
 package fs2.data.csv
 
-import cats.data.NonEmptyList
+import cats.syntax.all._
+
+/** A CSV row
+  * 
+  * Operations on columns can be performed using 0-based indices
+  */
+case class Row(values: List[String],
+                                            line: Option[Long] = None) {
+
+  /** Number of cells in the row. */
+  def size: Int = values.size
+
+  /**
+    * Set the line number for this row.
+    */
+  def withLine(line: Option[Long]): Row =
+    copy(line = line)
+
+  /** Returns the content of the cell at `idx` if it exists.
+    * Returns `None` if `idx` is out of row bounds.
+    * An empty cell value results in `Some("")`.
+    */
+  def at(idx: Int): Option[String] =
+    values.get(idx.toLong)
+
+  /** Returns the decoded content of the cell at `idx`.
+    * Fails if the index doesn't exist or cannot be decoded
+    * to the expected type.
+    */
+  def asAt[T](idx: Int)(implicit decoder: CellDecoder[T]): DecoderResult[T] =
+    values.get(idx.toLong) match {
+      case Some(v) => decoder(v)
+      case None    => Left(new DecoderError(s"unknown index $idx"))
+    }
+
+  /** Returns the decoded content of the cell at `idx` wrapped in Some if the cell is non-empty, None otherwise.
+    * Fails if the index doesn't exist or cannot be decoded
+    * to the expected type.
+    */
+  @deprecated(message =
+                "Use `RowF.asOptionalAt` instead, as it gives more flexibility and has the same default behavior.",
+              since = "fs2-data 1.7.0")
+  def asNonEmptyAt[T](idx: Int)(implicit decoder: CellDecoder[T]): DecoderResult[Option[T]] =
+    asOptionalAt(idx)
+
+  /** Returns the decoded content of the cell at `idx` wrapped in Some if the cell is non-empty, `None` otherwise.
+    * The meaning of _empty_ can be tuned by setting providing a custom `isEmpty` predicate (by default, matches the empty string).
+    * In case the index does not exist, the `missing` parameter defines the behavior (by default, it faile)
+    * Fails if the index cannot be decoded to the expected type.
+    */
+  def asOptionalAt[T](
+      idx: Int,
+      missing: Int => DecoderResult[Option[T]] = (idx: Int) => Left(new DecoderError(s"unknown index $idx")),
+      isEmpty: String => Boolean = _.isEmpty)(implicit decoder: CellDecoder[T]): DecoderResult[Option[T]] =
+    values.get(idx.toLong) match {
+      case Some(v) if isEmpty(v) => Right(None)
+      case Some(v)               => decoder.apply(v).map(Some(_))
+      case None                  => missing(idx)
+    }
+
+  /** Modifies the cell content at the given `idx` using the function `f`.
+    */
+  def modifyAt(idx: Int)(f: String => String): Row =
+    if (idx < 0 || idx >= values.size)
+      this
+    else
+      new Row(values.zipWithIndex.map { case (cell, i) => if (i === idx) f(cell) else cell })
+
+  /** Returns the row with the cell at `idx` modified to `value`. */
+  def updatedAt(idx: Int, value: String): Row =
+    modifyAt(idx)(_ => value)
+
+  /** Returns the row without the cell at the given `idx`.
+    * If the resulting row is empty, returns `None`.
+    */
+  def deleteAt(idx: Int): Row =
+    if (idx < 0 || idx >= values.size) {
+      this
+    } else {
+      val (before, after) = values.toList.splitAt(idx)
+      this.copy(values = before ++ after.tail)
+    }
+
+}
 
 object Row {
-  def apply(values: NonEmptyList[String], line: Option[Long] = None): Row = new Row(values, None, line)
-  def unapply(arg: Row): Some[NonEmptyList[String]] = Some(arg.values)
+  def unapply(row: Row): Option[List[String]] =
+    Some(row.values)
 }
diff --git a/csv/shared/src/main/scala/fs2/data/csv/RowDecoder.scala b/csv/shared/src/main/scala/fs2/data/csv/RowDecoder.scala
index d6956bc05..90fcd92b6 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/RowDecoder.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/RowDecoder.scala
@@ -16,19 +16,143 @@
 
 package fs2.data.csv
 
+import cats._
+import cats.syntax.all._
+
+import scala.annotation.tailrec
+
 /** Describes how a row can be decoded to the given type.
   *
-  * `RowDecoder` provides convenient methods such as `map`, `emap`, or `flatMap`
+  * `RowDecoderF` provides convenient methods such as `map`, `emap`, or `flatMap`
   * to build new decoders out of more basic one.
   *
-  * Actually, `RowDecoder` has a [[https://typelevel.org/cats/api/cats/MonadError.html cats `MonadError`]]
+  * Actually, `RowDecoderF` has a [[https://typelevel.org/cats/api/cats/MonadError.html cats `MonadError`]]
   * instance. To get the full power of it, import `cats.syntax.all._`.
   */
+@FunctionalInterface sealed trait RowDecoder[T] {
+  import RowDecoder.instance
+  def apply(row: Row): DecoderResult[T]
+
+  /** Map the parsed value.
+    * @param f the mapping function
+    * @tparam T2 the result type
+    * @return a row decoder reading the mapped type
+    */
+  def map[T2](f: T => T2): RowDecoder[T2] =
+    instance(row => apply(row).map(f))
+
+  /** Map the parsed value to a new decoder, which in turn will be applied to
+    * the parsed value.
+    * @param f the mapping function
+    * @tparam T2 the result type
+    * @return a row decoder reading the mapped type
+    */
+  def flatMap[T2](f: T => RowDecoder[T2]): RowDecoder[T2] =
+    instance(row => apply(row).flatMap(f(_)(row)))
+
+  /** Map the parsed value, potentially failing.
+    * @param f the mapping function
+    * @tparam T2 the result type
+    * @return a row decoder reading the mapped type
+    */
+  def emap[T2](f: T => DecoderResult[T2]): RowDecoder[T2] =
+    instance(row => apply(row).flatMap(f))
+
+  /** Fail-over. If this decoder fails, try the supplied other decoder.
+    * @param cd the fail-over decoder
+    * @tparam TT the return type
+    * @return a decoder combining this and the other decoder
+    */
+  def or[TT >: T](cd: => RowDecoder[TT]): RowDecoder[TT] =
+    instance { row =>
+      apply(row) match {
+        case Left(_)      => cd(row)
+        case r @ Right(_) => r.leftCast[DecoderError]
+      }
+    }
+  /** Similar to [[or]], but return the result as an Either signaling which row decoder succeeded. Allows for parsing
+    * an unrelated type in case of failure.
+    * @param cd the alternative decoder
+    * @tparam B the type the alternative decoder returns
+    * @return a decoder combining both decoders
+    */
+  def either[B](cd: RowDecoder[B]): RowDecoder[Either[T, B]] =
+    instance { row =>
+      apply(row) match {
+        case Left(_) =>
+          cd(row) match {
+            case l @ Left(_)  => l.rightCast[Either[T, B]]
+            case r @ Right(_) => r.leftCast[T].asRight
+          }
+        case Right(value) => Right(Left(value))
+      }
+    }
+}
+
+object RowDecoder extends ExportedRowDecoders {
 
-object RowDecoder {
   @inline
   def apply[T: RowDecoder]: RowDecoder[T] = implicitly[RowDecoder[T]]
 
   @inline
-  def instance[T](f: Row => DecoderResult[T]): RowDecoder[T] = row => f(row)
+  private [csv] def instance[T](f: Row => DecoderResult[T]): RowDecoder[T] =
+    new RowDecoder[T] {
+      def apply(row: Row) = f(row)
+    }
+
+
+  implicit def identityRowDecoder: RowDecoder[Row] = instance(_.asRight)
+
+  implicit def RowDecoderInstances
+      : MonadError[RowDecoder[*], DecoderError] with SemigroupK[RowDecoder[*]] =
+    new MonadError[RowDecoder[*], DecoderError] with SemigroupK[RowDecoder[*]] {
+      override def map[A, B](fa: RowDecoder[A])(f: A => B): RowDecoder[B] =
+        fa.map(f)
+
+      def flatMap[A, B](fa: RowDecoder[A])(f: A => RowDecoder[B]): RowDecoder[B] =
+        fa.flatMap(f)
+
+      def handleErrorWith[A](fa: RowDecoder[A])(
+          f: DecoderError => RowDecoder[A]): RowDecoder[A] =
+        instance(row => fa(row).leftFlatMap(f(_)(row)))
+
+      def pure[A](x: A): RowDecoder[A] =
+        instance(_ => Right(x))
+
+      def raiseError[A](e: DecoderError): RowDecoder[A] =
+        instance(_ => Left(e))
+
+      def tailRecM[A, B](a: A)(f: A => RowDecoder[Either[A, B]]): RowDecoder[B] = {
+        @tailrec
+        def step(row: Row, a: A): DecoderResult[B] =
+          f(a)(row) match {
+            case left @ Left(_)          => left.rightCast[B]
+            case Right(Left(a))          => step(row, a)
+            case Right(right @ Right(_)) => right.leftCast[DecoderError]
+          }
+        instance(row => step(row, a))
+      }
+
+      def combineK[A](x: RowDecoder[A], y: RowDecoder[A]): RowDecoder[A] = x or y
+    }
+
+  implicit val toListRowDecoder: RowDecoder[List[String]] =
+    instance(_.values.toList.asRight)
+
+  implicit val toNelRowDecoder: RowDecoder[List[String]] =
+    instance(_.values.asRight)
+
+  implicit def decodeResultRowDecoder[T](implicit dec: RowDecoder[T]): RowDecoder[DecoderResult[T]] =
+    instance(r => Right(dec(r)))
+
+  def asAt[T: CellDecoder](idx: Int): RowDecoder[T] =
+    instance { (row: Row) =>
+      row.values.toList.lift(idx).toRight {
+        new DecoderError(s"unknown index $idx")
+      }.flatMap(CellDecoder[T].apply)
+    }
+}
+
+trait ExportedRowDecoders {
+  implicit def exportedRowDecoders[A](implicit exported: Exported[RowDecoder[A]]): RowDecoder[A] = exported.instance
 }
diff --git a/csv/shared/src/main/scala/fs2/data/csv/RowDecoderF.scala b/csv/shared/src/main/scala/fs2/data/csv/RowDecoderF.scala
deleted file mode 100644
index 220d297cf..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/RowDecoderF.scala
+++ /dev/null
@@ -1,154 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import cats._
-import cats.data.{NonEmptyList, NonEmptyMap}
-import cats.syntax.all._
-
-import scala.annotation.tailrec
-
-/** Describes how a row can be decoded to the given type.
-  *
-  * `RowDecoderF` provides convenient methods such as `map`, `emap`, or `flatMap`
-  * to build new decoders out of more basic one.
-  *
-  * Actually, `RowDecoderF` has a [[https://typelevel.org/cats/api/cats/MonadError.html cats `MonadError`]]
-  * instance. To get the full power of it, import `cats.syntax.all._`.
-  */
-@FunctionalInterface trait RowDecoderF[H[+a] <: Option[a], T, Header] {
-  def apply(row: RowF[H, Header]): DecoderResult[T]
-
-  /** Map the parsed value.
-    * @param f the mapping function
-    * @tparam T2 the result type
-    * @return a row decoder reading the mapped type
-    */
-  def map[T2](f: T => T2): RowDecoderF[H, T2, Header] =
-    row => apply(row).map(f)
-
-  /** Map the parsed value to a new decoder, which in turn will be applied to
-    * the parsed value.
-    * @param f the mapping function
-    * @tparam T2 the result type
-    * @return a row decoder reading the mapped type
-    */
-  def flatMap[T2](f: T => RowDecoderF[H, T2, Header]): RowDecoderF[H, T2, Header] =
-    row => apply(row).flatMap(f(_)(row))
-
-  /** Map the parsed value, potentially failing.
-    * @param f the mapping function
-    * @tparam T2 the result type
-    * @return a row decoder reading the mapped type
-    */
-  def emap[T2](f: T => DecoderResult[T2]): RowDecoderF[H, T2, Header] =
-    row => apply(row).flatMap(f)
-
-  /** Fail-over. If this decoder fails, try the supplied other decoder.
-    * @param cd the fail-over decoder
-    * @tparam TT the return type
-    * @return a decoder combining this and the other decoder
-    */
-  def or[TT >: T](cd: => RowDecoderF[H, TT, Header]): RowDecoderF[H, TT, Header] =
-    row =>
-      apply(row) match {
-        case Left(_)      => cd(row)
-        case r @ Right(_) => r.leftCast[DecoderError]
-      }
-
-  /** Similar to [[or]], but return the result as an Either signaling which row decoder succeeded. Allows for parsing
-    * an unrelated type in case of failure.
-    * @param cd the alternative decoder
-    * @tparam B the type the alternative decoder returns
-    * @return a decoder combining both decoders
-    */
-  def either[B](cd: RowDecoderF[H, B, Header]): RowDecoderF[H, Either[T, B], Header] =
-    row =>
-      apply(row) match {
-        case Left(_) =>
-          cd(row) match {
-            case l @ Left(_)  => l.rightCast[Either[T, B]]
-            case r @ Right(_) => r.leftCast[T].asRight
-          }
-        case Right(value) => Right(Left(value))
-      }
-}
-
-object RowDecoderF extends ExportedRowDecoderFs {
-
-  implicit def identityRowDecoderF[H[+a] <: Option[a], Header]: RowDecoderF[H, RowF[H, Header], Header] = _.asRight
-
-  implicit def RowDecoderFInstances[H[+a] <: Option[a], Header]
-      : MonadError[RowDecoderF[H, *, Header], DecoderError] with SemigroupK[RowDecoderF[H, *, Header]] =
-    new MonadError[RowDecoderF[H, *, Header], DecoderError] with SemigroupK[RowDecoderF[H, *, Header]] {
-      override def map[A, B](fa: RowDecoderF[H, A, Header])(f: A => B): RowDecoderF[H, B, Header] =
-        fa.map(f)
-
-      def flatMap[A, B](fa: RowDecoderF[H, A, Header])(f: A => RowDecoderF[H, B, Header]): RowDecoderF[H, B, Header] =
-        fa.flatMap(f)
-
-      def handleErrorWith[A](fa: RowDecoderF[H, A, Header])(
-          f: DecoderError => RowDecoderF[H, A, Header]): RowDecoderF[H, A, Header] =
-        row => fa(row).leftFlatMap(f(_)(row))
-
-      def pure[A](x: A): RowDecoderF[H, A, Header] =
-        _ => Right(x)
-
-      def raiseError[A](e: DecoderError): RowDecoderF[H, A, Header] =
-        _ => Left(e)
-
-      def tailRecM[A, B](a: A)(f: A => RowDecoderF[H, Either[A, B], Header]): RowDecoderF[H, B, Header] = {
-        @tailrec
-        def step(row: RowF[H, Header], a: A): DecoderResult[B] =
-          f(a)(row) match {
-            case left @ Left(_)          => left.rightCast[B]
-            case Right(Left(a))          => step(row, a)
-            case Right(right @ Right(_)) => right.leftCast[DecoderError]
-          }
-        row => step(row, a)
-      }
-
-      def combineK[A](x: RowDecoderF[H, A, Header], y: RowDecoderF[H, A, Header]): RowDecoderF[H, A, Header] = x or y
-    }
-
-  implicit def toMapCsvRowDecoder[Header]: CsvRowDecoder[Map[Header, String], Header] =
-    CsvRowDecoder.instance(_.toMap.asRight)
-
-  implicit def toNonEmptyMapCsvRowDecoder[Header: Order]: CsvRowDecoder[NonEmptyMap[Header, String], Header] =
-    CsvRowDecoder.instance(_.toNonEmptyMap.asRight)
-
-  implicit val toListRowDecoder: RowDecoder[List[String]] =
-    RowDecoder.instance(_.values.toList.asRight)
-
-  implicit val toNelRowDecoder: RowDecoder[NonEmptyList[String]] =
-    RowDecoder.instance(_.values.asRight)
-
-  // the following two can't be unified va RowDecoderF because type inference fails us
-  implicit def decodeResultCsvRowDecoder[Header, T](implicit
-      dec: CsvRowDecoder[T, Header]): CsvRowDecoder[DecoderResult[T], Header] =
-    r => Right(dec(r))
-
-  implicit def decodeResultRowDecoder[T](implicit dec: RowDecoder[T]): RowDecoder[DecoderResult[T]] =
-    r => Right(dec(r))
-}
-
-trait ExportedRowDecoderFs {
-  implicit def exportedRowDecoders[A](implicit exported: Exported[RowDecoder[A]]): RowDecoder[A] = exported.instance
-
-  implicit def exportedCsvRowDecoders[A](implicit
-      exported: Exported[CsvRowDecoder[A, String]]): CsvRowDecoder[A, String] = exported.instance
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/RowEncoder.scala b/csv/shared/src/main/scala/fs2/data/csv/RowEncoder.scala
index 4f0009e10..c2825c819 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/RowEncoder.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/RowEncoder.scala
@@ -16,13 +16,47 @@
 
 package fs2.data.csv
 
-import cats.data.NonEmptyList
+import cats._
+import cats.syntax.all._
 
-object RowEncoder {
+import scala.annotation.implicitNotFound
 
+/** Describes how a row can be encoded from a value of the given type.
+  */
+@implicitNotFound(
+  "No implicit RowEncoderF[ found for type ${T}.\nYou can define one using RowEncoderF[.instance, by calling contramap on another RowEncoderF[ or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val csvRowEncoder: RowEncoderF[[${T}] = deriveRowEncoderF[\nMake sure to have instances of CellEncoder for every member type in scope.\n")
+@FunctionalInterface trait RowEncoder[T] {
+  def apply(elem: T): Row
+
+  def contramap[B](f: B => T): RowEncoder[B] = elem => apply(f(elem))
+}
+
+object RowEncoder extends ExportedRowEncoders {
   @inline
   def apply[T: RowEncoder]: RowEncoder[T] = implicitly[RowEncoder[T]]
 
   @inline
-  def instance[T](f: T => NonEmptyList[String]): RowEncoder[T] = (t: T) => Row(f(t))
+  def instance[T](f: T => List[String]): RowEncoder[T] = (t: T) => Row(f(t))
+
+  implicit def identityRowEncoder: RowEncoder[Row] = identity
+
+  implicit def RowEncoder: ContravariantMonoidal[RowEncoder] =
+    new ContravariantMonoidal[RowEncoder] {
+      override def product[A, B](fa: RowEncoder[A], fb: RowEncoder[B]): RowEncoder[(A, B)] =
+        (fa, fb).tupled
+
+      override def unit: RowEncoder[Unit] =
+        _ => Row(Nil, None)
+
+      override def contramap[A, B](fa: RowEncoder[A])(f: B => A): RowEncoder[B] =
+        fa.contramap(f)
+    }
+
+  implicit val fromNelRowEncoder: RowEncoder[List[String]] =
+    instance[List[String]](r => r)
+}
+
+trait ExportedRowEncoders {
+  implicit def exportedRowEncoder[A](implicit exported: Exported[RowEncoder[A]]): RowEncoder[A] = exported.instance
+
 }
diff --git a/csv/shared/src/main/scala/fs2/data/csv/RowEncoderF.scala b/csv/shared/src/main/scala/fs2/data/csv/RowEncoderF.scala
deleted file mode 100644
index cc1f07888..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/RowEncoderF.scala
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import cats._
-import cats.data.{NonEmptyList, NonEmptyMap}
-
-import scala.annotation.{implicitNotFound, unused}
-
-/** Describes how a row can be encoded from a value of the given type.
-  */
-@implicitNotFound(
-  "No implicit RowEncoderF[H,  found for type ${T}.\nYou can define one using RowEncoderF[H, .instance, by calling contramap on another RowEncoderF[H,  or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val csvRowEncoder: RowEncoderF[H, [${T}] = deriveRowEncoderF[H, \nMake sure to have instances of CellEncoder for every member type in scope.\n")
-@FunctionalInterface trait RowEncoderF[H[+a] <: Option[a], T, Header] {
-  def apply(elem: T): RowF[H, Header]
-
-  def contramap[B](f: B => T): RowEncoderF[H, B, Header] = elem => apply(f(elem))
-}
-
-object RowEncoderF extends ExportedRowEncoderFs {
-
-  implicit def identityRowEncoderF[H[+a] <: Option[a], Header]: RowEncoderF[H, RowF[H, Header], Header] = identity
-
-  implicit def RowEncoderF[H[+a] <: Option[a], Header]: Contravariant[RowEncoderF[H, *, Header]] =
-    new Contravariant[RowEncoderF[H, *, Header]] {
-      override def contramap[A, B](fa: RowEncoderF[H, A, Header])(f: B => A): RowEncoderF[H, B, Header] =
-        fa.contramap(f)
-    }
-
-  @inline
-  def apply[H[+a] <: Option[a], T: RowEncoderF[H, *, Header], Header]: RowEncoderF[H, T, Header] =
-    implicitly[RowEncoderF[H, T, Header]]
-
-  @inline
-  def instance[H[+a] <: Option[a], T, Header](f: T => RowF[H, Header]): RowEncoderF[H, T, Header] = f(_)
-
-  @deprecated("retained for bincompat", "1.6.0")
-  private[csv] implicit def fromNonEmptyMapCsvRowEncoder[Header](implicit
-      @unused ev1: Order[Header]): CsvRowEncoder[NonEmptyMap[Header, String], Header] =
-    CsvRowEncoder.instance(_.toNel)
-
-  implicit def fromNonEmptyMapCsvRowEncoder[Header]: CsvRowEncoder[NonEmptyMap[Header, String], Header] =
-    CsvRowEncoder.instance(_.toNel)
-
-  implicit val fromNelRowEncoder: RowEncoder[NonEmptyList[String]] =
-    RowEncoder.instance(r => r)
-}
-
-trait ExportedRowEncoderFs {
-  implicit def exportedRowEncoder[A](implicit exported: Exported[RowEncoder[A]]): RowEncoder[A] = exported.instance
-
-  implicit def exportedCsvRowEncoderF[A](implicit
-      exported: Exported[CsvRowEncoder[A, String]]): CsvRowEncoder[A, String] = exported.instance
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/RowF.scala b/csv/shared/src/main/scala/fs2/data/csv/RowF.scala
deleted file mode 100644
index ad13e4f6e..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/RowF.scala
+++ /dev/null
@@ -1,242 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import cats._
-import cats.data._
-import cats.syntax.all._
-
-import scala.annotation.{nowarn, unused}
-
-/** A CSV row with or without headers. The presence of headers is encoded via the first type param
-  * which is a subtype of [[scala.Option]]. By preserving this information in types, it's possible to define
-  * [[Row]] and [[CsvRow]] aliases as if they were plain case classes while keeping the code DRY.
-  *
-  * Operations on columns can always be performed using 0-based indices and additionally using a specified header value
-  * if headers are present (and this fact statically known).
-  *
-  * '''Note:''' the following invariant holds when using this class: `values` and `headers` have the same size if headers are present.
-  */
-case class RowF[H[+a] <: Option[a], Header](values: NonEmptyList[String],
-                                            headers: H[NonEmptyList[Header]],
-                                            line: Option[Long] = None) {
-
-  /** Number of cells in the row. */
-  def size: Int = values.size
-
-  /**
-    * Set the line number for this row.
-    */
-  def withLine(line: Option[Long]): RowF[H, Header] =
-    copy(line = line)
-
-  /** Returns the content of the cell at `idx` if it exists.
-    * Returns `None` if `idx` is out of row bounds.
-    * An empty cell value results in `Some("")`.
-    */
-  def at(idx: Int): Option[String] =
-    values.get(idx.toLong)
-
-  /** Returns the decoded content of the cell at `idx`.
-    * Fails if the index doesn't exist or cannot be decoded
-    * to the expected type.
-    */
-  def asAt[T](idx: Int)(implicit decoder: CellDecoder[T]): DecoderResult[T] =
-    values.get(idx.toLong) match {
-      case Some(v) => decoder(v)
-      case None    => Left(new DecoderError(s"unknown index $idx"))
-    }
-
-  /** Returns the decoded content of the cell at `idx` wrapped in Some if the cell is non-empty, None otherwise.
-    * Fails if the index doesn't exist or cannot be decoded
-    * to the expected type.
-    */
-  @deprecated(message =
-                "Use `RowF.asOptionalAt` instead, as it gives more flexibility and has the same default behavior.",
-              since = "fs2-data 1.7.0")
-  def asNonEmptyAt[T](idx: Int)(implicit decoder: CellDecoder[T]): DecoderResult[Option[T]] =
-    asOptionalAt(idx)
-
-  /** Returns the decoded content of the cell at `idx` wrapped in Some if the cell is non-empty, `None` otherwise.
-    * The meaning of _empty_ can be tuned by setting providing a custom `isEmpty` predicate (by default, matches the empty string).
-    * In case the index does not exist, the `missing` parameter defines the behavior (by default, it faile)
-    * Fails if the index cannot be decoded to the expected type.
-    */
-  def asOptionalAt[T](
-      idx: Int,
-      missing: Int => DecoderResult[Option[T]] = (idx: Int) => Left(new DecoderError(s"unknown index $idx")),
-      isEmpty: String => Boolean = _.isEmpty)(implicit decoder: CellDecoder[T]): DecoderResult[Option[T]] =
-    values.get(idx.toLong) match {
-      case Some(v) if isEmpty(v) => Right(None)
-      case Some(v)               => decoder.apply(v).map(Some(_))
-      case None                  => missing(idx)
-    }
-
-  /** Modifies the cell content at the given `idx` using the function `f`.
-    */
-  def modifyAt(idx: Int)(f: String => String): RowF[H, Header] =
-    if (idx < 0 || idx >= values.size)
-      this
-    else
-      new RowF[H, Header](values.zipWithIndex.map { case (cell, i) => if (i === idx) f(cell) else cell }, headers)
-
-  /** Modifies the cell content at the given `header` using the function `f`.
-    *
-    * **Note:** Only the first occurrence of the values with the given header
-    * will be modified. It shouldn't be a problem in the general case as headers
-    * should not be duplicated.
-    */
-  def modify(header: Header)(f: String => String)(implicit hasHeaders: HasHeaders[H, Header]): CsvRow[Header] =
-    hasHeaders(modifyAt(headers.get.toList.indexOf(header))(f))
-
-  /** Returns the row with the cell at `idx` modified to `value`. */
-  def updatedAt(idx: Int, value: String): RowF[H, Header] =
-    modifyAt(idx)(_ => value)
-
-  /** Returns the row with the cell at `header` modified to `value`.
-    *
-    * **Note:** Only the first occurrence of the values with the given header
-    * will be modified. It shouldn't be a problem in the general case as headers
-    * should not be duplicated.
-    */
-  def updated(header: Header, value: String)(implicit hasHeaders: HasHeaders[H, Header]): CsvRow[Header] =
-    hasHeaders(updatedAt(headers.get.toList.indexOf(header), value))
-
-  /** Returns the row with the cell at `header` modified to `value`.
-    * If the header wasn't present in the row, it is added to the end of the fields.
-    *
-    * **Note:** Only the first occurrence of the values with the given header
-    * will be modified. It shouldn't be a problem in the general case as headers
-    * should not be duplicated.
-    */
-  def set(header: Header, value: String)(implicit hasHeaders: HasHeaders[H, Header]): CsvRow[Header] = {
-    val idx = headers.get.toList.indexOf(header)
-    if (idx < 0)
-      hasHeaders(new RowF(values :+ value, headers.map(_ :+ header).asInstanceOf[H[NonEmptyList[Header]]]))
-    else
-      hasHeaders(updatedAt(idx, value))
-  }
-
-  /** Returns the row without the cell at the given `idx`.
-    * If the resulting row is empty, returns `None`.
-    */
-  def deleteAt(idx: Int): Option[RowF[H, Header]] =
-    if (idx < 0 || idx >= values.size) {
-      Some(this)
-    } else {
-      val (before, after) = values.toList.splitAt(idx)
-      val nh = htraverse(headers) { headers =>
-        val (h1, h2) = headers.toList.splitAt(idx)
-        NonEmptyList.fromList(h1 ::: h2.tail)
-      }
-      (NonEmptyList.fromList(before ++ after.tail), nh).mapN(new RowF[H, Header](_, _))
-    }
-
-  /** Returns the row without the cell at the given `header`.
-    * If the resulting row is empty, returns `None`.
-    *
-    * **Note:** Only the first occurrence of the values with the given header
-    * will be deleted. It shouldn't be a problem in the general case as headers
-    * should not be duplicated.
-    */
-  def delete(header: Header)(implicit hasHeaders: HasHeaders[H, Header]): Option[CsvRow[Header]] =
-    deleteAt(headers.get.toList.indexOf(header)).map(hasHeaders)
-
-  /** Returns the content of the cell at `header` if it exists.
-    * Returns `None` if `header` does not exist for the row.
-    * An empty cell value results in `Some("")`.
-    */
-  def apply(header: Header)(implicit @unused hasHeaders: HasHeaders[H, Header]): Option[String] =
-    byHeader.get(header): @nowarn("msg=HasHeaders")
-
-  /** Returns the decoded content of the cell at `header`.
-    * Fails if the field doesn't exist or cannot be decoded
-    * to the expected type.
-    */
-  def as[T](
-      header: Header)(implicit @unused hasHeaders: HasHeaders[H, Header], decoder: CellDecoder[T]): DecoderResult[T] =
-    (byHeader: @nowarn("msg=HasHeaders")).get(header) match {
-      case Some(v) => decoder(v)
-      case None    => Left(new DecoderError(s"unknown field $header"))
-    }
-
-  /** Returns the decoded content of the cell at `header` wrapped in Some if the cell is non-empty, None otherwise.
-    * Fails if the field doesn't exist or cannot be decoded
-    * to the expected type.
-    */
-  @deprecated(message =
-                "Use `RowF.asOptional` instead, as it gives more flexibility and has the same default behavior.",
-              since = "fs2-data 1.7.0")
-  def asNonEmpty[T](
-      header: Header)(implicit hasHeaders: HasHeaders[H, Header], decoder: CellDecoder[T]): DecoderResult[Option[T]] =
-    asOptional(header)
-
-  /** Returns the decoded content of the cell at `header` wrapped in Some if the cell is non-empty, `None` otherwise.
-    * The meaning of _empty_ can be tuned by setting providing a custom `isEmpty` predicate (by default, matches the empty string).
-    * In case the field does not exist, the `missing` parameter defines the behavior (by default, it faile)
-    * Fails if the index cannot be decoded to the expected type.
-    */
-  def asOptional[T](header: Header,
-                    missing: Header => DecoderResult[Option[T]] = (header: Header) =>
-                      Left(new DecoderError(s"unknown field $header")),
-                    isEmpty: String => Boolean = _.isEmpty)(implicit
-      @unused hasHeaders: HasHeaders[H, Header],
-      decoder: CellDecoder[T]): DecoderResult[Option[T]] =
-    (byHeader: @nowarn("msg=HasHeaders")).get(header) match {
-      case Some(v) if isEmpty(v) => Right(None)
-      case Some(v)               => decoder(v).map(Some(_))
-      case None                  => missing(header)
-    }
-
-  /** Returns a representation of this row as Map from headers to corresponding cell values.
-    */
-  def toMap(implicit @unused hasHeaders: HasHeaders[H, Header]): Map[Header, String] =
-    byHeader: @nowarn("msg=HasHeaders")
-
-  /** Returns a representation of this row as NonEmptyMap from headers to corresponding cell values.
-    */
-  def toNonEmptyMap(implicit
-      @unused hasHeaders: HasHeaders[H, Header],
-      order: Order[Header]): NonEmptyMap[Header, String] =
-    headers.get.zip(values).toNem
-
-  /** Drop all headers (if any).
-    * @return a row without headers, but same values
-    */
-  def dropHeaders: Row = Row(values)
-
-  // let's cache this to avoid recomputing it for every call to `as` or similar method
-  // the `Option.get` call is safe since this field is only called in a context where a `HasHeaders`
-  // instance is provided, meaning that the `Option` is `Some`
-  // of course using a lazy val prevents us to make this constraint statistically checked, but
-  // the gain is significant enough to allow for this local unsafety
-  @deprecated("Have you checked that you have a `HasHeaders` instance in scope?", "fs2-data 1.5.0")
-  private lazy val byHeader: Map[Header, String] =
-    headers.get.toList.zip(values.toList).toMap
-
-  // Like Traverse[Option], but preserves the H type
-  private def htraverse[G[_]: Applicative, A, B](h: H[A])(f: A => G[B]): G[H[B]] = h match {
-    case Some(a) => f(a).map(Some(_)).asInstanceOf[G[H[B]]]
-    case _       => Applicative[G].pure(None).asInstanceOf[G[H[B]]]
-  }
-}
-
-object RowF {
-  implicit object functor extends Functor[CsvRow[*]] {
-    override def map[A, B](fa: CsvRow[A])(f: A => B): CsvRow[B] = fa.copy(headers = Some(fa.headers.get.map(f)))
-  }
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/WriteableHeader.scala b/csv/shared/src/main/scala/fs2/data/csv/WriteableHeader.scala
index d70c565aa..ea4bc38c0 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/WriteableHeader.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/WriteableHeader.scala
@@ -17,10 +17,9 @@
 package fs2.data.csv
 
 import cats.Contravariant
-import cats.data.NonEmptyList
 
 trait WriteableHeader[Header] {
-  def apply(headers: NonEmptyList[Header]): NonEmptyList[String]
+  def apply(headers: List[Header]): List[String]
 
   def contramap[B](f: B => Header): WriteableHeader[B] = headers => apply(headers.map(f))
 }
@@ -37,7 +36,7 @@ object WriteableHeader {
     _.map(encode)
 
   implicit object StringWriteableHeader extends WriteableHeader[String] {
-    def apply(names: NonEmptyList[String]): NonEmptyList[String] = names
+    def apply(names: List[String]): List[String] = names
   }
 
   implicit object WriteableHeaderInstances extends Contravariant[WriteableHeader] {
diff --git a/csv/shared/src/main/scala/fs2/data/csv/exceptions.scala b/csv/shared/src/main/scala/fs2/data/csv/exceptions.scala
index bd5f0ccd8..c68ac590e 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/exceptions.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/exceptions.scala
@@ -27,7 +27,7 @@ class DecoderError(msg: String, override val line: Option[Long] = None, inner: T
 }
 
 class HeaderError(msg: String, override val line: Option[Long] = None, inner: Throwable = null)
-    extends CsvException(msg, line, inner) {
+    extends DecoderError(msg, line, inner) {
   override def withLine(line: Option[Long]): HeaderError = new HeaderError(msg, line, inner)
 }
 
diff --git a/csv/shared/src/main/scala/fs2/data/csv/internals/CsvRowParser.scala b/csv/shared/src/main/scala/fs2/data/csv/internals/CsvRowParser.scala
deleted file mode 100644
index 0f75024ee..000000000
--- a/csv/shared/src/main/scala/fs2/data/csv/internals/CsvRowParser.scala
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2
-package data
-package csv
-package internals
-
-import cats.syntax.all._
-
-private[csv] object CsvRowParser {
-
-  def pipe[F[_], Header](implicit
-      F: RaiseThrowable[F],
-      Header: ParseableHeader[Header]
-  ): Pipe[F, Row, CsvRow[Header]] =
-    _.through(pipeAttempt[F, Header]).rethrow
-
-  /** Like `pipe` except that instead of failing the stream on parse errors, it emits `Left` elements for bad rows */
-  def pipeAttempt[F[_], Header](implicit
-      Header: ParseableHeader[Header]): Pipe[F, Row, Either[CsvException, CsvRow[Header]]] =
-    _.pull.uncons1
-      .flatMap {
-        case Some((firstRow, tail)) =>
-          Header(firstRow.values) match {
-            case Left(error) => Pull.output1(Left(error))
-            case Right(headers) if headers.length =!= firstRow.values.length =>
-              val error = new HeaderError(
-                s"Got ${headers.length} headers, but ${firstRow.values.length} columns. Both numbers must match!",
-                firstRow.line)
-              Pull.output1(Left(error))
-            case Right(headers) =>
-              tail
-                .map(CsvRow.liftRow(headers))
-                .pull
-                .echo
-          }
-        case None => Pull.done
-      }
-      .stream
-
-}
diff --git a/csv/shared/src/main/scala/fs2/data/csv/internals/RowParser.scala b/csv/shared/src/main/scala/fs2/data/csv/internals/RowParser.scala
index c7662cab5..2b6ce8fe4 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/internals/RowParser.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/internals/RowParser.scala
@@ -21,8 +21,6 @@ package internals
 
 import text._
 
-import cats.data.{State => _, _}
-
 import scala.collection.immutable.VectorBuilder
 
 private[csv] object RowParser {
@@ -47,14 +45,14 @@ private[csv] object RowParser {
             state match {
               case State.BeginningOfField =>
                 if (tail.nonEmpty)
-                  Pull.output1(Row(NonEmptyList("", tail).reverse, Some(line))) >> Pull.done
+                  Pull.output1(Row(("" :: tail).reverse, Some(line))) >> Pull.done
                 else
                   Pull.done
               case State.InUnquoted | State.InQuotedSeenQuote | State.ExpectNewLine =>
-                Pull.output1(Row(NonEmptyList(currentField.result(), tail).reverse, Some(line))) >> Pull.done
+                Pull.output1(Row((currentField.result() :: tail).reverse, Some(line))) >> Pull.done
               case State.InUnquotedSeenCr =>
                 Pull.output1(
-                  Row(NonEmptyList(currentField.append('\r').result(), tail).reverse, Some(line))) >> Pull.done
+                  Row((currentField.append('\r').result() :: tail).reverse, Some(line))) >> Pull.done
               case State.InQuoted =>
                 Pull.raiseError[F](new CsvException("unexpected end of input", Some(line)))
             }
@@ -87,7 +85,7 @@ private[csv] object RowParser {
                    Nil,
                    State.BeginningOfField,
                    line + 1,
-                   chunkAcc += Row(NonEmptyList(field, tail).reverse, Some(line)))
+                   chunkAcc += Row((field :: tail).reverse, Some(line)))
             } else if (c == '\r') {
               rows(T.advance(context), currentField, tail, State.ExpectNewLine, line, chunkAcc)
             } else {
@@ -104,7 +102,7 @@ private[csv] object RowParser {
                    Nil,
                    State.BeginningOfField,
                    line + 1,
-                   chunkAcc += Row(NonEmptyList(field, tail).reverse, Some(line)))
+                   chunkAcc += Row((field :: tail).reverse, Some(line)))
             } else {
               // this is an error
               Pull.output(Chunk.from(chunkAcc.result())) >> Pull.raiseError[F](
@@ -125,7 +123,7 @@ private[csv] object RowParser {
                      Nil,
                      State.BeginningOfField,
                      line + 1,
-                     chunkAcc += Row(NonEmptyList("", tail).reverse, Some(line)))
+                     chunkAcc += Row(("" :: tail).reverse, Some(line)))
               } else {
                 rows(T.advance(context), currentField, Nil, State.BeginningOfField, line + 1, chunkAcc)
               }
@@ -149,7 +147,7 @@ private[csv] object RowParser {
                    Nil,
                    State.BeginningOfField,
                    line + 1,
-                   chunkAcc += Row(NonEmptyList(field, tail).reverse, Some(line)))
+                   chunkAcc += Row((field :: tail).reverse, Some(line)))
             } else if (c == '\r') {
               rows(T.advance(context), currentField, tail, State.InUnquotedSeenCr, line, chunkAcc)
             } else {
@@ -165,7 +163,7 @@ private[csv] object RowParser {
                    Nil,
                    State.BeginningOfField,
                    line + 1,
-                   chunkAcc += Row(NonEmptyList(field, tail).reverse, Some(line)))
+                   chunkAcc += Row((field :: tail).reverse, Some(line)))
             } else {
               currentField.append('\r')
               if (c == separator) {
diff --git a/csv/shared/src/main/scala/fs2/data/csv/package.scala b/csv/shared/src/main/scala/fs2/data/csv/package.scala
index 13c250f05..6688ebfe7 100644
--- a/csv/shared/src/main/scala/fs2/data/csv/package.scala
+++ b/csv/shared/src/main/scala/fs2/data/csv/package.scala
@@ -23,25 +23,10 @@ import csv.internals._
 import cats.data._
 import cats.syntax.all._
 
-import scala.annotation.{implicitNotFound, unused}
 import scala.annotation.nowarn
 
 package object csv {
 
-  /** Higher kinded version of [[scala.None]]. Ignores the type param.
-    */
-  type NoneF[+A] = None.type
-
-  /** A CSV row without headers.
-    */
-  type Row = RowF[NoneF, Nothing]
-
-  /** A CSV row with headers, that can be used to access the cell values.
-    *
-    * '''Note:''' the following invariant holds when using this class: `values` and `headers` have the same size.
-    */
-  type CsvRow[Header] = RowF[Some, Header]
-
   type HeaderResult[T] = Either[HeaderError, NonEmptyList[T]]
 
   type DecoderResult[T] = Either[DecoderError, T]
@@ -54,33 +39,6 @@ package object csv {
     * Actually, `RowDecoder` has a [[https://typelevel.org/cats/api/cats/MonadError.html cats `MonadError`]]
     * instance. To get the full power of it, import `cats.syntax.all._`.
     */
-  @implicitNotFound(
-    "No implicit RowDecoder found for type ${T}.\nYou can define one using RowDecoder.instance, by calling map on another RowDecoder or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val rowDecoder: RowDecoder[${T}] = deriveRowDecoder\nMake sure to have instances of CellDecoder for every member type in scope.\n")
-  type RowDecoder[T] = RowDecoderF[NoneF, T, Nothing]
-
-  /** Describes how a row can be encoded from a value of the given type.
-    */
-  @implicitNotFound(
-    "No implicit RowEncoder found for type ${T}.\nYou can define one using RowEncoder.instance, by calling contramap on another RowEncoder or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val rowEncoder: RowEncoder[${T}] = deriveRowEncoder\nMake sure to have instances of CellEncoder for every member type in scope.\n")
-  type RowEncoder[T] = RowEncoderF[NoneF, T, Nothing]
-
-  /** Describes how a row can be decoded to the given type.
-    *
-    * `CsvRowDecoder` provides convenient methods such as `map`, `emap`, or `flatMap`
-    * to build new decoders out of more basic one.
-    *
-    * Actually, `CsvRowDecoder` has a [[https://typelevel.org/cats/api/cats/MonadError.html cats `MonadError`]]
-    * instance. To get the full power of it, import `cats.syntax.all._`.
-    */
-  @implicitNotFound(
-    "No implicit CsvRowDecoder found for type ${T}.\nYou can define one using CsvRowDecoder.instance, by calling map on another CsvRowDecoder or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val csvRowDecoder: CsvRowDecoder[${T}] = deriveCsvRowDecoder\nMake sure to have instances of CellDecoder for every member type in scope.\n")
-  type CsvRowDecoder[T, Header] = RowDecoderF[Some, T, Header]
-
-  /** Describes how a row can be encoded from a value of the given type.
-    */
-  @implicitNotFound(
-    "No implicit CsvRowEncoderF[H,  found for type ${T}.\nYou can define one using CsvRowEncoderF[H, .instance, by calling contramap on another CsvRowEncoderF[H,  or by using generic derivation for product types like case classes.\nFor that, add the fs2-data-csv-generic module to your dependencies and use either full-automatic derivation:\nimport fs2.data.csv.generic.auto._\nor the recommended semi-automatic derivation:\nimport fs2.data.csv.generic.semiauto._\nimplicit val csvRowEncoder: CsvRowEncoderF[H, [${T}] = deriveCsvRowEncoderF[H, \nMake sure to have instances of CellEncoder for every member type in scope.\n")
-  type CsvRowEncoder[T, Header] = RowEncoderF[Some, T, Header]
 
   @nowarn
   sealed trait QuoteHandling
@@ -143,41 +101,12 @@ package object csv {
 
   @nowarn
   class PartiallyAppliedDecodeUsingHeaders[T](val dummy: Boolean) extends AnyVal {
-    def apply[F[_], C, Header](separator: Char = ',', quoteHandling: QuoteHandling = QuoteHandling.RFCCompliant)(
+    def apply[F[_], C](separator: Char = ',', quoteHandling: QuoteHandling = QuoteHandling.RFCCompliant)(
         implicit
         F: RaiseThrowable[F],
         C: CharLikeChunks[F, C],
-        T: CsvRowDecoder[T, Header],
-        H: ParseableHeader[Header]): Pipe[F, C, T] =
-      lowlevel.rows(separator, quoteHandling) andThen lowlevel.headers andThen lowlevel.decodeRow
-  }
-
-  /** Decode a char-like stream (see [[fs2.data.text.CharLikeChunks]]) into a specified type.
-    *
-    * Scenarios:
-    * - If skipHeaders is false, then the file contains no headers.
-    * - If skipHeaders is true, then the headers in the file will be skipped.
-    *
-    * For both scenarios the file is assumed to be compliant with the set of headers given.
-    */
-  def decodeGivenHeaders[T]: PartiallyAppliedDecodeGivenHeaders[T] =
-    new PartiallyAppliedDecodeGivenHeaders(dummy = true)
-
-  @nowarn
-  class PartiallyAppliedDecodeGivenHeaders[T](val dummy: Boolean) extends AnyVal {
-    def apply[F[_], C, Header](headers: NonEmptyList[Header],
-                               skipHeaders: Boolean = false,
-                               separator: Char = ',',
-                               quoteHandling: QuoteHandling = QuoteHandling.RFCCompliant)(implicit
-        F: RaiseThrowable[F],
-        C: CharLikeChunks[F, C],
-        T: CsvRowDecoder[T, Header]): Pipe[F, C, T] = {
-      if (skipHeaders)
-        lowlevel.rows(separator, quoteHandling) andThen lowlevel.skipHeaders andThen
-          lowlevel.withHeaders(headers) andThen lowlevel.decodeRow
-      else
-        lowlevel.rows(separator, quoteHandling) andThen lowlevel.withHeaders(headers) andThen lowlevel.decodeRow
-    }
+        T: CsvRowDecoder[T]): Pipe[F, C, T] =
+      lowlevel.rows(separator, quoteHandling) andThen lowlevel.decodeRow[F, T]
   }
 
   /** Encode a specified type into a CSV that contains no headers. */
@@ -203,7 +132,7 @@ package object csv {
 
   @nowarn
   class PartiallyAppliedEncodeGivenHeaders[T](val dummy: Boolean) extends AnyVal {
-    def apply[F[_], Header](headers: NonEmptyList[Header],
+    def apply[F[_], Header](headers: List[Header],
                             fullRows: Boolean = false,
                             separator: Char = ',',
                             newline: String = "\n",
@@ -218,21 +147,20 @@ package object csv {
   }
 
   /** Encode a specified type into a CSV that contains the headers determined by encoding the first element. Empty if input is. */
-  def encodeUsingFirstHeaders[T]: PartiallyAppliedEncodeUsingFirstHeaders[T] =
+  def encodeUsingHeaders[T]: PartiallyAppliedEncodeUsingFirstHeaders[T] =
     new PartiallyAppliedEncodeUsingFirstHeaders(dummy = true)
 
   @nowarn
   class PartiallyAppliedEncodeUsingFirstHeaders[T](val dummy: Boolean) extends AnyVal {
-    def apply[F[_], Header](fullRows: Boolean = false,
-                            separator: Char = ',',
-                            newline: String = "\n",
-                            escape: EscapeMode = EscapeMode.Auto)(implicit
-        T: CsvRowEncoder[T, Header],
-        H: WriteableHeader[Header]): Pipe[F, T, String] = {
+    def apply[F[_]](fullRows: Boolean = false,
+                    separator: Char = ',',
+                    newline: String = "\n",
+                    escape: EscapeMode = EscapeMode.Auto)(implicit
+        T: CsvRowEncoder[T]): Pipe[F, T, String] = {
       val stringPipe =
         if (fullRows) lowlevel.toRowStrings[F](separator, newline, escape)
         else lowlevel.toStrings[F](separator, newline, escape)
-      lowlevel.encodeRow[F, Header, T] andThen lowlevel.encodeRowWithFirstHeaders[F, Header] andThen stringPipe
+      lowlevel.encodeRowWithHeaders[F, T] andThen stringPipe
     }
   }
 
@@ -256,33 +184,6 @@ package object csv {
         T: CharLikeChunks[F, T]): Pipe[F, T, Row] =
       RowParser.pipe[F, T](separator, quoteHandling)
 
-    /** Transforms a stream of raw CSV rows into parsed CSV rows with headers. */
-    def headers[F[_], Header](implicit
-        F: RaiseThrowable[F],
-        Header: ParseableHeader[Header]): Pipe[F, Row, CsvRow[Header]] =
-      CsvRowParser.pipe[F, Header]
-
-    /** Transforms a stream of raw CSV rows into parsed CSV rows with headers, with failures at the element level instead of failing the stream */
-    def headersAttempt[F[_], Header](implicit
-        Header: ParseableHeader[Header]): Pipe[F, Row, Either[CsvException, CsvRow[Header]]] =
-      CsvRowParser.pipeAttempt[F, Header]
-
-    // left here for bincompat
-    private[csv] def headersAttempt[F[_], Header](implicit
-        @unused F: RaiseThrowable[F],
-        Header: ParseableHeader[Header]): Pipe[F, Row, Either[CsvException, CsvRow[Header]]] =
-      CsvRowParser.pipeAttempt[F, Header]
-
-    /** Transforms a stream of raw CSV rows into parsed CSV rows with given headers. */
-    def withHeaders[F[_], Header](headers: NonEmptyList[Header])(implicit
-        F: RaiseThrowable[F]): Pipe[F, Row, CsvRow[Header]] =
-      _.map(CsvRow.liftRow(headers)).rethrow
-
-    /** Transforms a stream of raw CSV rows into parsed CSV rows with given headers. */
-    def attemptWithHeaders[F[_], Header](
-        headers: NonEmptyList[Header]): Pipe[F, Row, Either[CsvException, CsvRow[Header]]] =
-      _.map(CsvRow.liftRow(headers))
-
     /** Transforms a stream of raw CSV rows into rows. */
     def noHeaders[F[_]]: Pipe[F, Row, Row] = identity
 
@@ -299,35 +200,44 @@ package object csv {
       _.map(R(_))
 
     /** Decodes [[CsvRow]]s (with headers) into a specified type using a suitable [[CsvRowDecoder]]. */
-    def decodeRow[F[_], Header, R](implicit
+    def decodeRow[F[_], R](implicit
         F: RaiseThrowable[F],
-        R: CsvRowDecoder[R, Header]): Pipe[F, CsvRow[Header], R] =
-      _.map(R(_)).rethrow
+        R: CsvRowDecoder[R]): Pipe[F, Row, R] =
+      s => s.through(attemptDecodeRow[F, R]).rethrow
 
     /** Decodes [[CsvRow]]s (with headers) into a specified type using a suitable [[CsvRowDecoder]], but signal errors as values. */
-    def attemptDecodeRow[F[_], Header, R](implicit
-        R: CsvRowDecoder[R, Header]): Pipe[F, CsvRow[Header], DecoderResult[R]] =
-      _.map(R(_))
-
-    /** Decodes [[CsvRow]]s (with headers) into a specified type using a suitable [[CsvRowDecoder]], but signal errors as values from both header as well as rows. */
-    def attemptFlatMapDecodeRow[F[_], Header, R](implicit R: CsvRowDecoder[R, Header])
-        : Stream[F, Either[CsvException, CsvRow[Header]]] => Stream[F, Either[CsvException, R]] = {
-      _.map(_.flatMap(R(_)))
-    }
+    def attemptDecodeRow[F[_], R](implicit
+        R: CsvRowDecoder[R]): Pipe[F, Row, DecoderResult[R]] =
+      (_: Stream[F, Row]).pull.uncons1
+        .flatMap {
+          case Some((headers, tail)) =>
+            tail.through(attemptDecodeRowWithHeaders[F, R](headers.values)).pull.echo
+          case None => Pull.done
+        }.stream
+
+    def attemptDecodeRowWithHeaders[F[_], R](headers: List[String])(implicit
+        R: CsvRowDecoder[R]): Pipe[F, Row, DecoderResult[R]] =
+      stream => 
+        R(headers) match {
+          case Left(err) =>
+            Stream(Left(err))
+          case Right(p) =>
+            stream.through(attemptDecode(p))
+        }
 
     /** Encode a given type into CSV rows using a set of explicitly given headers. */
-    def writeWithHeaders[F[_], Header](headers: NonEmptyList[Header])(implicit
-        H: WriteableHeader[Header]): Pipe[F, Row, NonEmptyList[String]] =
+    def writeWithHeaders[F[_], Header](headers: List[Header])(implicit
+        H: WriteableHeader[Header]): Pipe[F, Row, List[String]] =
       Stream(H(headers)) ++ _.map(_.values)
 
     /** Encode a given type into CSV rows without headers. */
-    def writeWithoutHeaders[F[_]]: Pipe[F, Row, NonEmptyList[String]] =
+    def writeWithoutHeaders[F[_]]: Pipe[F, Row, List[String]] =
       _.map(_.values)
 
     /** Serialize a CSV row to Strings. No guarantees are given on how the resulting Strings are cut. */
     def toStrings[F[_]](separator: Char = ',',
                         newline: String = "\n",
-                        escape: EscapeMode = EscapeMode.Auto): Pipe[F, NonEmptyList[String], String] = {
+                        escape: EscapeMode = EscapeMode.Auto): Pipe[F, List[String], String] = {
       _.flatMap(nel =>
         Stream
           .emits(nel.toList)
@@ -338,7 +248,7 @@ package object csv {
     /** Serialize a CSV row to Strings. Guaranteed to emit one String per CSV row (= one line if no quoted newlines are contained in the value). */
     def toRowStrings[F[_]](separator: Char = ',',
                            newline: String = "\n",
-                           escape: EscapeMode = EscapeMode.Auto): Pipe[F, NonEmptyList[String], String] = {
+                           escape: EscapeMode = EscapeMode.Auto): Pipe[F, List[String], String] = {
       // explicit Show avoids mapping the NEL before
       val showColumn: Show[String] =
         Show.show(RowWriter.encodeColumn(separator, escape))
@@ -350,26 +260,19 @@ package object csv {
       _.map(R(_))
 
     /** Encode a given type into CSV rows with headers. */
-    def encodeRow[F[_], Header, R](implicit R: CsvRowEncoder[R, Header]): Pipe[F, R, CsvRow[Header]] =
-      _.map(R(_))
+    def encodeRow[F[_], R](implicit R: CsvRowEncoder[R]): Pipe[F, R, Row] =
+      _.map(R.encoder(_))
 
     /** Encode a given type into CSV row with headers taken from the first element.
-      * If the input stream is empty, the output is as well.
+      * If the input stream is empty, the output contains only the headers
       */
-    def encodeRowWithFirstHeaders[F[_], Header](implicit
-        H: WriteableHeader[Header]): Pipe[F, CsvRow[Header], NonEmptyList[String]] =
-      _.pull.peek1
-        .flatMap {
-          case Some((CsvRow(_, headers), stream)) =>
-            Pull.output1(H(headers)) >> stream.map(_.values).pull.echo
-          case None => Pull.done
-        }
-        .stream
-
+    def encodeRowWithHeaders[F[_], R: CsvRowEncoder]: Pipe[F, R, List[String]] =
+      stream => 
+        Stream(CsvRowEncoder[R].headers) ++
+          stream.through(encodeRow[F, R]).map(_.values)
   }
 
   object lenient {
-
     /** Decode a char-like stream (see [[fs2.data.text.CharLikeChunks]]) into a specified type, with failures at the
       * element level instead of failing the stream.
       *
@@ -383,19 +286,19 @@ package object csv {
       new PartiallyAppliedDecodeAttemptGivenHeaders[T](dummy = true)
 
     class PartiallyAppliedDecodeAttemptGivenHeaders[T](val dummy: Boolean) extends AnyVal {
-      def apply[F[_], C, Header](headers: NonEmptyList[Header],
-                                 skipHeaders: Boolean = false,
-                                 separator: Char = ',',
-                                 quoteHandling: QuoteHandling = QuoteHandling.RFCCompliant)(implicit
+      def apply[F[_], C](headers: List[String],
+                         skipHeaders: Boolean = false,
+                         separator: Char = ',',
+                         quoteHandling: QuoteHandling = QuoteHandling.RFCCompliant)(implicit
           F: RaiseThrowable[F],
           C: CharLikeChunks[F, C],
-          T: CsvRowDecoder[T, Header]): Pipe[F, C, Either[CsvException, T]] = {
+          T: CsvRowDecoder[T]): Pipe[F, C, Either[CsvException, T]] = {
         if (skipHeaders)
           lowlevel.rows(separator, quoteHandling) andThen lowlevel.skipHeaders andThen
-            lowlevel.attemptWithHeaders(headers) andThen lowlevel.attemptFlatMapDecodeRow
+            lowlevel.attemptDecodeRowWithHeaders(headers)
         else
-          lowlevel.rows(separator, quoteHandling) andThen lowlevel.attemptWithHeaders(
-            headers) andThen lowlevel.attemptFlatMapDecodeRow
+          lowlevel.rows(separator, quoteHandling) andThen lowlevel.attemptDecodeRowWithHeaders(
+            headers)
       }
     }
 
@@ -412,10 +315,8 @@ package object csv {
           implicit
           F: RaiseThrowable[F],
           C: CharLikeChunks[F, C],
-          T: CsvRowDecoder[T, Header],
-          H: ParseableHeader[Header]): Pipe[F, C, Either[CsvException, T]] = {
-        lowlevel.rows(separator, quoteHandling) andThen lowlevel.headersAttempt(
-          H) andThen lowlevel.attemptFlatMapDecodeRow
+          T: CsvRowDecoder[T]): Pipe[F, C, Either[CsvException, T]] = {
+        lowlevel.rows(separator, quoteHandling) andThen lowlevel.attemptDecodeRow[F, T]
       }
     }
 
diff --git a/csv/shared/src/test/scala/fs2/data/csv/CsvExceptionSpec.scala b/csv/shared/src/test/scala/fs2/data/csv/CsvExceptionSpec.scala
index 332e2d64a..e86bd5ac2 100644
--- a/csv/shared/src/test/scala/fs2/data/csv/CsvExceptionSpec.scala
+++ b/csv/shared/src/test/scala/fs2/data/csv/CsvExceptionSpec.scala
@@ -16,7 +16,6 @@
 
 package fs2.data.csv
 
-import cats.data.NonEmptyList
 import fs2.{Fallible, Stream}
 
 import weaver._
@@ -33,8 +32,8 @@ object CsvExceptionSpec extends SimpleIOSuite {
 
     expect(stream.compile.toList match {
       case Right(
-            List(Right(Row(NonEmptyList("1", List("2", "3")))),
-                 Right(Row(NonEmptyList("a", List("b", "c")))),
+            List(Right(Row(List("1", "2", "3"))),
+                 Right(Row(List("a", "b", "c"))),
                  Left(e: CsvException))) =>
         e.line.contains(3L) // check that we have the correct line number here
       case _ => false
diff --git a/csv/shared/src/test/scala/fs2/data/csv/CsvParserTest.scala b/csv/shared/src/test/scala/fs2/data/csv/CsvParserTest.scala
index 9507aa414..5c0a9ca9b 100644
--- a/csv/shared/src/test/scala/fs2/data/csv/CsvParserTest.scala
+++ b/csv/shared/src/test/scala/fs2/data/csv/CsvParserTest.scala
@@ -17,7 +17,6 @@
 package fs2.data.csv
 
 import cats.Eq
-import cats.data.NonEmptyList
 import io.circe.parser.parse
 import fs2._
 import fs2.io.file.{Files, Flags, Path}
@@ -33,27 +32,36 @@ object CsvParserTest extends SimpleIOSuite {
   case class Test(i: Int = 0, s: String, j: Option[Int])
   case class TestData(name: String, age: Int, description: String)
 
-  implicit val testDataCsvRowDecoder: CsvRowDecoder[TestData, String] = (row: RowF[Some, String]) => {
+  case class CsvRow(toMap: Map[String, String])
+  object CsvRow {
+    implicit val decoder: CsvRowDecoder[CsvRow] =
+      CsvRowDecoder.instance {
+        (headers: List[String]) =>
+          Right { RowDecoder.instance { (row: Row) =>
+            Right(CsvRow(headers.zip(row.values).toList.toMap))
+          }}
+      }
+    def fromListHeaders(l: List[(String, String)]): Option[CsvRow] =
+      Some(CsvRow(l.toMap))
+  }
+
+  implicit val testDataCsvRowDecoder: CsvRowDecoder[TestData] = 
     (
-      row.as[String]("name"),
-      row.as[Int]("age"),
-      row.as[String]("description")
+      CsvRowDecoder.as[String]("name"),
+      CsvRowDecoder.as[Int]("age"),
+      CsvRowDecoder.as[String]("description")
     ).mapN { case (name, age, description) =>
       TestData(name, age, description)
     }
-  }
 
-  implicit val testDataRowDecoder: RowDecoder[TestData] = new RowDecoder[TestData] {
-    override def apply(row: RowF[NoneF, Nothing]): DecoderResult[TestData] = {
+  implicit val testDataRowDecoder: RowDecoder[TestData] =
       (
-        row.asAt[String](0),
-        row.asAt[Int](1),
-        row.asAt[String](2)
+        RowDecoder.asAt[String](0),
+        RowDecoder.asAt[Int](1),
+        RowDecoder.asAt[String](2)
       ).mapN { case (name, age, description) =>
         TestData(name, age, description)
       }
-    }
-  }
 
   implicit val decoderErrorEq: Eq[CsvException] = Eq.by(_.toString)
   implicit val decoderTestDataEq: Eq[TestData] = Eq.fromUniversalEquals
@@ -82,7 +90,7 @@ object CsvParserTest extends SimpleIOSuite {
       .evalMap { case (path, expected) =>
         Files[IO]
           .readAll(path, 1024, Flags.Read)
-          .through(decodeUsingHeaders[CsvRow[String]]())
+          .through(decodeUsingHeaders[CsvRow]())
           .compile
           .toList
           .map(_.map(_.toMap))
@@ -96,13 +104,19 @@ object CsvParserTest extends SimpleIOSuite {
     allExpected
       .evalTap { case (path, _) => log.info(path.fileName.toString) }
       .evalMap { case (_, expected) =>
+        val headers = expected.head.keys.toList
+        implicit val dec = 
+          CsvRowEncoder.instance[CsvRow](
+            headers,
+            RowEncoder.instance(row => headers.map(row.toMap))
+          )
         Stream
           .emits(expected)
           .map(m => CsvRow.fromListHeaders(m.toList))
           .unNone
-          .through[IO, String](encodeUsingFirstHeaders[CsvRow[String]]())
+          .through[IO, String](encodeUsingHeaders[CsvRow]())
           .covary[IO]
-          .through(decodeUsingHeaders[CsvRow[String]]())
+          .through(decodeUsingHeaders[CsvRow]())
           .compile
           .toList
           .map(_.map(_.toMap))
@@ -131,7 +145,7 @@ object CsvParserTest extends SimpleIOSuite {
     Stream
       .emit(content)
       .covary[IO]
-      .through(decodeUsingHeaders[CsvRow[String]](',', QuoteHandling.Literal))
+      .through(decodeUsingHeaders[CsvRow](',', QuoteHandling.Literal))
       .compile
       .toList
       .map(_.map(_.toMap))
@@ -178,9 +192,6 @@ object CsvParserTest extends SimpleIOSuite {
 
     val expected = List(
       Left(new DecoderError("unknown field name", None)),
-      Left(new HeaderSizeError(3, 2, Some(3L))),
-      Left(new DecoderError("unknown field name", None)),
-      Left(new HeaderSizeError(3, 2, Some(5L)))
     )
 
     val stream = Stream
@@ -218,7 +229,7 @@ object CsvParserTest extends SimpleIOSuite {
       .through(
         lenient.attemptDecodeGivenHeaders[TestData](separator = ',',
                                                     skipHeaders = true,
-                                                    headers = NonEmptyList.of("name", "age", "description")))
+                                                    headers = List("name", "age", "description")))
       .compile
       .toList
 
@@ -249,7 +260,7 @@ object CsvParserTest extends SimpleIOSuite {
       .through(
         lenient.attemptDecodeGivenHeaders[TestData](separator = ',',
                                                     skipHeaders = false,
-                                                    headers = NonEmptyList.of("name", "age", "description")))
+                                                    headers = List("name", "age", "description")))
       .compile
       .toList
 
@@ -317,16 +328,9 @@ object CsvParserTest extends SimpleIOSuite {
   }
 
   pureTest("should fail if a required string field is missing") {
-    val row = CsvRow
-      .unsafe(
-        NonEmptyList.of("12", "3"),
-        NonEmptyList.of("i", "j")
-      )
-      .withLine(Some(2))
-
-    testDataCsvRowDecoder(row) match {
+    testDataCsvRowDecoder(List("i", "j")) match {
       case Left(error) => expect.eql(error.getMessage, "unknown field name")
-      case Right(x)    => failure(s"Stream succeeded with value $x")
+      case Right(x) => failure(s"Stream succeeded with value $x")
     }
   }
 
@@ -359,35 +363,6 @@ object CsvParserTest extends SimpleIOSuite {
     }
   }
 
-  pureTest("Parser should return only errors as values when using attemptDecodeUsingHeaders with wrong header") {
-    val content =
-      """name1,age,description
-        |John Doe,47,description 1
-        |Jane Doe,50
-        |Bob Smith,80,description 2
-        |Alice Grey,78
-        |""".stripMargin
-
-    val expected = List(
-      Left(new DecoderError("unknown field name", None)),
-      Left(new HeaderSizeError(3, 2, Some(3L))),
-      Left(new DecoderError("unknown field name", None)),
-      Left(new HeaderSizeError(3, 2, Some(5L)))
-    )
-
-    val stream = Stream
-      .emit(content)
-      .covary[Fallible]
-      .through(lenient.attemptDecodeUsingHeaders[TestData](','))
-      .compile
-      .toList
-
-    stream match {
-      case Right(actual) => expect.eql(expected, actual)
-      case Left(x)       => failure(s"Stream failed with value $x")
-    }
-  }
-
   pureTest("Parser should return all decoder results as values when using attemptDecodeGivenHeaders") {
     val content =
       """name-should-not-be-used,age-should-not-be-used,description-should-not-be-used
@@ -410,7 +385,7 @@ object CsvParserTest extends SimpleIOSuite {
       .through(
         lenient.attemptDecodeGivenHeaders[TestData](separator = ',',
                                                     skipHeaders = true,
-                                                    headers = NonEmptyList.of("name", "age", "description")))
+                                                    headers = List("name", "age", "description")))
       .compile
       .toList
 
@@ -441,7 +416,7 @@ object CsvParserTest extends SimpleIOSuite {
       .through(
         lenient.attemptDecodeGivenHeaders[TestData](separator = ',',
                                                     skipHeaders = false,
-                                                    headers = NonEmptyList.of("name", "age", "description")))
+                                                    headers = List("name", "age", "description")))
       .compile
       .toList
 
diff --git a/csv/shared/src/test/scala/fs2/data/csv/LineNumberPreservationTest.scala b/csv/shared/src/test/scala/fs2/data/csv/LineNumberPreservationTest.scala
deleted file mode 100644
index 381025ade..000000000
--- a/csv/shared/src/test/scala/fs2/data/csv/LineNumberPreservationTest.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import cats.effect._
-import cats.syntax.all._
-import fs2._
-import fs2.data._
-import weaver._
-
-object LineNumberPreservationTest extends SimpleIOSuite {
-
-  test("lowlevel.headers pipe preserves line numbers") {
-    Stream("""h1,h2,h3
-             |a,b,c
-             |d,e,f
-             |g,h,i
-             |""".stripMargin)
-      .covary[IO]
-      .through(csv.lowlevel.rows())
-      .through(csv.lowlevel.headers[IO, String])
-      .zipWithIndex
-      .map { case (row, idx) => expect.eql((idx + 2).some, row.line) } // idx +1 for headers, +1 as lines are 1-based
-      .compile
-      .foldMonoid
-  }
-
-}
diff --git a/csv/shared/src/test/scala/fs2/data/csv/RowDecoderFTest.scala b/csv/shared/src/test/scala/fs2/data/csv/RowDecoderTest.scala
similarity index 82%
rename from csv/shared/src/test/scala/fs2/data/csv/RowDecoderFTest.scala
rename to csv/shared/src/test/scala/fs2/data/csv/RowDecoderTest.scala
index 0924151c4..22243db79 100644
--- a/csv/shared/src/test/scala/fs2/data/csv/RowDecoderFTest.scala
+++ b/csv/shared/src/test/scala/fs2/data/csv/RowDecoderTest.scala
@@ -20,8 +20,10 @@ package csv
 
 import cats.effect.IO
 import weaver._
+import cats.syntax.all._
+import fs2.data.csv.lenient.attemptDecodeUsingHeaders
 
-object RowDecoderFTest extends SimpleIOSuite {
+object RowDecoderTest extends SimpleIOSuite {
 
   case class TwoNumbers(a: Int, b: Int)
 
@@ -32,12 +34,13 @@ object RowDecoderFTest extends SimpleIOSuite {
     } yield TwoNumbers(a, b)
   }
 
-  implicit val csvRowDecoder: CsvRowDecoder[TwoNumbers, String] = CsvRowDecoder.instance { row =>
-    for {
-      a <- row.as[Int]("a")
-      b <- row.as[Int]("b")
-    } yield TwoNumbers(a, b)
-  }
+  implicit val csvRowDecoder: CsvRowDecoder[TwoNumbers] =
+    (
+      CsvRowDecoder.as[Int]("a"),
+      CsvRowDecoder.as[Int]("b")
+    ).mapN(
+      TwoNumbers(_, _)
+    )
 
   test("can parse CSV with rows that do not convert by an attempting RowDecoder") {
     val rows = List(
@@ -62,12 +65,11 @@ object RowDecoderFTest extends SimpleIOSuite {
     )
     Stream
       .emits[IO, String](rows)
-      .through(decodeUsingHeaders[DecoderResult[TwoNumbers]]())
+      .through(attemptDecodeUsingHeaders[TwoNumbers]())
       .compile
       .toList
       .map { result =>
         expect(result.head.isRight) and expect(result.tail.head.isLeft)
       }
   }
-
 }
diff --git a/csv/shared/src/test/scala/fs2/data/csv/RowFTest.scala b/csv/shared/src/test/scala/fs2/data/csv/RowFTest.scala
deleted file mode 100644
index 3f18f2b4e..000000000
--- a/csv/shared/src/test/scala/fs2/data/csv/RowFTest.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Copyright 2024 fs2-data Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package fs2.data.csv
-
-import weaver._
-import cats.data.NonEmptyList
-
-object RowFTest extends SimpleIOSuite {
-
-  pureTest("RowF.set should change the value at existing header") {
-    val row = CsvRow.unsafe(NonEmptyList.of("1", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    expect.eql(NonEmptyList.of("1", "4", "3"), row.set("b", "4").values)
-  }
-
-  pureTest("RowF.set should add the value at end of row in case of missing header") {
-    val row = CsvRow.unsafe(NonEmptyList.of("1", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    val extended = row.set("d", "4")
-    expect.eql(NonEmptyList.of("1", "2", "3", "4"), extended.values) and
-      expect.eql(NonEmptyList.of("a", "b", "c", "d"), extended.headers.get)
-  }
-
-  pureTest("CsvRow.asOptional should return None for empty cells") {
-    val row = CsvRow.unsafe(NonEmptyList.of("", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    expect(row.asOptional[Int]("a").contains(None))
-  }
-
-  pureTest("CsvRow.asOptional should return None for missing cells") {
-    val row = CsvRow.unsafe(NonEmptyList.of("", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    expect(row.asOptional[Int]("d", missing = _ => Right(None)).contains(None))
-  }
-
-  pureTest("CsvRow.asOptional should return None for values considered empty") {
-    val row = CsvRow.unsafe(NonEmptyList.of("N/A", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    expect(row.asOptional[Int]("a", isEmpty = _ == "N/A").contains(None))
-  }
-
-  pureTest("CsvRow.asOptional should return decoded value for non-empty cells") {
-    val row = CsvRow.unsafe(NonEmptyList.of("", "2", "3"), NonEmptyList.of("a", "b", "c"))
-    expect(row.asOptional[Int]("b").contains(Some(2)))
-  }
-
-  pureTest("Row.asOptionalAt should return None for empty cells") {
-    val row = Row(NonEmptyList.of("", "2", "3"))
-    expect(row.asOptionalAt[Int](0).contains(None))
-  }
-
-  pureTest("CsvRow.asOptionalAt should return None for missing cells") {
-    val row = Row(NonEmptyList.of("", "2", "3"))
-    expect(row.asOptionalAt[Int](7, missing = _ => Right(None)).contains(None))
-  }
-
-  pureTest("CsvRow.asOptionalAt should return None for values considered empty") {
-    val row = Row(NonEmptyList.of("N/A", "2", "3"))
-    expect(row.asOptionalAt[Int](0, isEmpty = _ == "N/A").contains(None))
-  }
-
-  pureTest("Row.asOptionalAt should return decoded value for non-empty cells") {
-    val row = Row(NonEmptyList.of("", "2", "3"))
-    expect(row.asOptionalAt[Int](1).contains(Some(2)))
-  }
-}
diff --git a/csv/shared/src/test/scala/fs2/data/csv/RowTest.scala b/csv/shared/src/test/scala/fs2/data/csv/RowTest.scala
new file mode 100644
index 000000000..7f9c011f2
--- /dev/null
+++ b/csv/shared/src/test/scala/fs2/data/csv/RowTest.scala
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2024 fs2-data Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fs2.data.csv
+
+import weaver._
+
+object RowTest extends SimpleIOSuite {
+
+  pureTest("Row.asOptionalAt should return None for empty cells") {
+    val row = new Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](0).contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return None for missing cells") {
+    val row = new Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](3, missing = _ => Right(None)).contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return None for values considered empty") {
+    val row = new Row(List("N/A", "2", "3"))
+    expect(row.asOptionalAt[Int](0, isEmpty = _ == "N/A").contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return decoded value for non-empty cells") {
+    val row = new Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](1).contains(Some(2)))
+  }
+
+  pureTest("Row.asOptionalAt should return None for empty cells") {
+    val row = Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](0).contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return None for missing cells") {
+    val row = Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](7, missing = _ => Right(None)).contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return None for values considered empty") {
+    val row = Row(List("N/A", "2", "3"))
+    expect(row.asOptionalAt[Int](0, isEmpty = _ == "N/A").contains(None))
+  }
+
+  pureTest("Row.asOptionalAt should return decoded value for non-empty cells") {
+    val row = Row(List("", "2", "3"))
+    expect(row.asOptionalAt[Int](1).contains(Some(2)))
+  }
+}