diff --git a/cwms-data-api/src/main/java/cwms/cda/ApiServlet.java b/cwms-data-api/src/main/java/cwms/cda/ApiServlet.java
index 7944c04b4b..bc4d6f2dfa 100644
--- a/cwms-data-api/src/main/java/cwms/cda/ApiServlet.java
+++ b/cwms-data-api/src/main/java/cwms/cda/ApiServlet.java
@@ -187,6 +187,8 @@
 import io.javalin.http.Handler;
 import io.javalin.http.HttpResponseException;
 import io.javalin.http.JavalinServlet;
+import cwms.cda.data.dto.csv.CwmsCsvDTO;
+import cwms.cda.formatters.csv.CsvExampleGenerator;
 import io.javalin.plugin.openapi.OpenApiOptions;
 import io.javalin.plugin.openapi.OpenApiPlugin;
 import io.swagger.v3.oas.models.Components;
@@ -194,6 +196,8 @@
 import io.swagger.v3.oas.models.Operation;
 import io.swagger.v3.oas.models.PathItem;
 import io.swagger.v3.oas.models.info.Info;
+import io.swagger.v3.oas.models.media.MediaType;
+import io.swagger.v3.oas.models.responses.ApiResponse;
 import io.swagger.v3.oas.models.security.SecurityRequirement;
 import io.swagger.v3.oas.models.servers.Server;
 import java.io.IOException;
@@ -224,6 +228,8 @@
 import org.jooq.exception.DataAccessException;
 import org.owasp.html.HtmlPolicyBuilder;
 import org.owasp.html.PolicyFactory;
+import io.github.classgraph.ClassGraph;
+import io.github.classgraph.ScanResult;
 
 
 /**
@@ -946,7 +952,37 @@ private void getOpenApiOptions(JavalinConfig config) {
                         });
                     }
             });
-                
+                Map<String, Class<? extends CwmsCsvDTO>> schemaToClass = new HashMap<>();
+                try (ScanResult scanResult = new ClassGraph()
+                        .acceptPackages("cwms.cda.data.dto")
+                        .scan()) {
+                    List<Class<CwmsCsvDTO>> csvDtoClasses = scanResult.getClassesImplementing(CwmsCsvDTO.class.getName())
+                            .loadClasses(CwmsCsvDTO.class);
+                    for (Class<? extends CwmsCsvDTO> clazz : csvDtoClasses) {
+                        schemaToClass.put(clazz.getSimpleName(), clazz);
+                    }
+                }
+                api.getPaths().values().forEach(pathItem -> {
+                    for (Operation op : pathItem.readOperations()) {
+                        if (op.getResponses() != null) {
+                            for (ApiResponse resp : op.getResponses().values()) {
+                                if (resp.getContent() != null && resp.getContent().containsKey(Formats.CSV)) {
+                                    MediaType csvMedia = resp.getContent().get(Formats.CSV);
+                                    if (csvMedia.getSchema() != null && csvMedia.getSchema().get$ref() != null) {
+                                        String ref = csvMedia.getSchema().get$ref();
+                                        String schemaName = ref.substring(ref.lastIndexOf('/') + 1);
+                                        @SuppressWarnings("unchecked")
+                                        Class<? extends CwmsCsvDTO<?>> dtoClass = (Class<? extends CwmsCsvDTO<?>>) schemaToClass.get(schemaName);
+
+                                        if (dtoClass != null) {
+                                            csvMedia.setExample(CsvExampleGenerator.getExample(dtoClass));
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                });
                 return api;
             })
             .defaultDocumentation(doc -> {
diff --git a/cwms-data-api/src/main/java/cwms/cda/api/Controllers.java b/cwms-data-api/src/main/java/cwms/cda/api/Controllers.java
index fef8f7322b..ed05a37267 100644
--- a/cwms-data-api/src/main/java/cwms/cda/api/Controllers.java
+++ b/cwms-data-api/src/main/java/cwms/cda/api/Controllers.java
@@ -40,7 +40,9 @@
 import io.javalin.core.validation.Validator;
 import io.javalin.http.Context;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -127,6 +129,8 @@ public final class Controllers {
     public static final String MATCH_NULL_PARENTS = "match-null-parents";
     public static final String ENTITY_ID = "entity-id";
     public static final String PARENT_ENTITY_ID = "parent-entity-id";
+    public static final String INCLUDE_METADATA_AS_CSV_COMMENTS = "include-metadata-as-comments";
+    public static final String INCLUDE_OPTIONAL_CSV_COLUMNS = "include-optional-csv-columns";
 
     public static final String CREATE_AS_LRTS = "create-as-lrts";
     public static final String STORE_RULE = "store-rule";
@@ -172,9 +176,8 @@ public final class Controllers {
     public static final String TS_IDS = "ts-ids";
 
     public static final String EXAMPLE_DATE = "2021-06-10T13:00:00-07:00";
-    public static final String DATE_FORMAT = "YYYY-MM-dd'T'hh:mm:ss[Z'['VV']']";
-    public static final String TIME_FORMAT_DESC = "The <a href=\"times.html\">format for this field</a> is ISO 8601 extended" +
-            ", with optional offset and timezone, i.e., '" + DATE_FORMAT + "', e.g., '" + EXAMPLE_DATE + "'."            ;
+    public static final String TIME_FORMAT_DESC = "The <a href=\"times.html\">format for this field</a> " +
+            "is ISO 8601 extended in UTC, e.g., 2026-06-18T19:42:00Z";
 
     public static final String INCLUDE_ASSIGNED = "include-assigned";
     public static final String ANY_MASK = "*";
@@ -215,6 +218,8 @@ public final class Controllers {
     public static final String AREA_UNIT = "area-unit";
     public static final String STATION_UNIT = "station-unit";
     public static final String STAGE_UNIT = "stage-unit";
+    public static final String DATE_FORMAT = "date-format";
+    public static final String DATE_FORMAT_PATTERN = "date-format-pattern";
     public static final String TRIM = "trim";
     public static final String DESIGNATOR = "designator";
     public static final String DESIGNATOR_MASK = "designator-mask";
diff --git a/cwms-data-api/src/main/java/cwms/cda/api/TimeSeriesController.java b/cwms-data-api/src/main/java/cwms/cda/api/TimeSeriesController.java
index 5a977d9616..ad5aefbfe6 100644
--- a/cwms-data-api/src/main/java/cwms/cda/api/TimeSeriesController.java
+++ b/cwms-data-api/src/main/java/cwms/cda/api/TimeSeriesController.java
@@ -21,6 +21,10 @@
 import cwms.cda.data.dao.VerticalDatum;
 import cwms.cda.data.dto.TimeSeries;
 import cwms.cda.formatters.ContentType;
+import cwms.cda.formatters.DateFormatResolver;
+import cwms.cda.formatters.DateFormat;
+import cwms.cda.formatters.csv.CsvConfiguration;
+import cwms.cda.data.dto.csv.TimeSeriesCsv;
 import cwms.cda.formatters.Formats;
 import cwms.cda.helpers.DateUtils;
 import io.javalin.apibuilder.CrudHandler;
@@ -112,7 +116,7 @@ public class TimeSeriesController implements CrudHandler {
     private final MetricRegistry metrics;
 
     private final Histogram requestResultSize;
-    private static final int DEFAULT_PAGE_SIZE = 500;
+    static final int DEFAULT_PAGE_SIZE = 500;
 
 
     public TimeSeriesController(MetricRegistry metrics) {
@@ -382,12 +386,26 @@ public void delete(@NotNull Context ctx, @NotNull String timeseries) {
                         + "identifies where in the request you are. This is an opaque"
                         + " value, and can be obtained from the 'next-page' value in "
                         + "the response. Deprecated, use " + PAGE + " instead."),
-                @OpenApiParam(name = PAGE_SIZE,
-                        type = Integer.class,
+                 @OpenApiParam(name = PAGE_SIZE, type = Integer.class,
                         description = "How many entries per page returned. "
-                                + "Default " + DEFAULT_PAGE_SIZE + ". Use 0 to return an empty values array, "
+                        + "For JSON/XML paging, this controls page size. "
+                        + "For CSV, this controls the internal fetch batch size used while streaming a single response. "
+                        + "CSV clients do not request subsequent pages. "
+                        + "Default " + DEFAULT_PAGE_SIZE +". Use 0 to return an empty values array, "
                                 + "or -1 to return the entire window in one response without a next-page cursor. "
-                                + "Values less than -1 are invalid.")
+                                + "Values less than -1 are invalid."),
+                @OpenApiParam(name = INCLUDE_METADATA_AS_CSV_COMMENTS, type = Boolean.class,
+                        description = "When true, include dataset metadata as csv header comments "
+                        + "prepended with # (default is false)."),
+                @OpenApiParam(name = INCLUDE_OPTIONAL_CSV_COLUMNS, type = Boolean.class,
+                        description = "When true, include optional columns (quality-code, data-entry-date) "
+                        + "in the CSV response (default is false)."),
+                @OpenApiParam(name = DATE_FORMAT,
+                        description = "Specifies the format of any dates in the response. "
+                        + "Default is ISO8601-Instant. Other possibilities are epoch-millis, ISO8601-Offset, "
+                        + "ISO8601-Local, date-only, and custom."),
+                @OpenApiParam(name = DATE_FORMAT_PATTERN,
+                        description = "When date-format is set to 'custom', this parameter specifies the date format pattern.")
             },
             responses = {
                 @OpenApiResponse(status = STATUS_200,
@@ -397,6 +415,7 @@ public void delete(@NotNull Context ctx, @NotNull String timeseries) {
                         @OpenApiContent(from = TimeSeries.class, type = Formats.XMLV2),
                         @OpenApiContent(from = TimeSeries.class, type = Formats.XML),
                         @OpenApiContent(from = TimeSeries.class, type = Formats.JSON),
+                        @OpenApiContent(from = TimeSeriesCsv.class, type= Formats.CSV),
                         @OpenApiContent(from = TimeSeries.class, type = ""),}),
                 @OpenApiResponse(status = STATUS_400, description = "Invalid parameter combination"),
                 @OpenApiResponse(status = STATUS_404, description = "The provided combination of "
@@ -444,8 +463,21 @@ public void getAll(@NotNull Context ctx) {
                     Integer.class, DEFAULT_PAGE_SIZE, metrics,
                     name(TimeSeriesController.class.getName(), GET_ALL)));
 
+            boolean includeMetadata = ctx.queryParamAsClass(INCLUDE_METADATA_AS_CSV_COMMENTS, Boolean.class)
+                    .getOrDefault(false);
+            boolean includeOptionalColumns = ctx.queryParamAsClass(INCLUDE_OPTIONAL_CSV_COLUMNS, Boolean.class)
+                    .getOrDefault(false);
+            String dateFormatParam = ctx.queryParam(DATE_FORMAT);
+            String dateFormatPattern = ctx.queryParam(DATE_FORMAT_PATTERN);
+
             String acceptHeader = ctx.header(Header.ACCEPT);
             ContentType contentType = Formats.parseHeaderAndQueryParm(acceptHeader, format, TimeSeries.class);
+            DateFormat dateFormat = DateFormatResolver.resolve(dateFormatParam, dateFormatPattern);
+            CsvConfiguration csvConfig = new CsvConfiguration.Builder()
+                    .withMetadataIncluded(includeMetadata)
+                    .withOptionalColumnsIncluded(includeOptionalColumns)
+                    .withDateFormat(dateFormat)
+                    .build();
 
             String results;
             String version = contentType.getParameters().get(VERSION);
@@ -471,6 +503,13 @@ public void getAll(@NotNull Context ctx) {
                         .withShouldTrim(trim.getOrDefault(true))
                         .withIncludeEntryDate(includeEntryDate)
                         .build();
+
+                // CSV: stream a single response; page-size is only internal batch size
+                if (Formats.CSV.equals(contentType.getType())) {
+                    streamCsv(ctx, csvConfig, pageSize, dao, requestParameters);
+                    return;
+                }
+
                 // Execute DAO call with a timeout so we can return a clearer message instead of a generic 500
                 int apiTimeoutMs = Integer.getInteger("cwms.cda.api.apiTimeoutMs", 45000);
                 CompletableFuture<TimeSeries> daoFuture = CompletableFuture.supplyAsync(
@@ -492,7 +531,7 @@ public void getAll(@NotNull Context ctx) {
                     throw unwrapExecutionException(ex);
                 }
 
-                if(datum != null) { //this will be null for non-elevation ts
+                if (datum != null) { //this will be null for non-elevation ts
                     // user has requested a specific vertical datum
                     VerticalDatum vd = VerticalDatum.valueOf(datum);  // the users request
                     ts = TimeSeriesVerticalDatumConverter.convertToVerticalDatum(ts, vd);
@@ -512,6 +551,24 @@ public void getAll(@NotNull Context ctx) {
                 ctx.header(Header.CONTENT_LENGTH, String.valueOf(bytes.length));
                 ctx.res.getOutputStream().write(bytes);
             } else {
+                String office = ctx.queryParam(OFFICE);
+
+                // CSV: stream a single response; page-size is only internal batch size
+                if (Formats.CSV.equals(contentType.getType())) {
+                    TimeSeriesRequestParameters requestParameters = new TimeSeriesRequestParameters.Builder()
+                            .withNames(names)
+                            .withOffice(office)
+                            .withUnits(units)
+                            .withBeginTime(beginZdt)
+                            .withEndTime(endZdt)
+                            .withShouldTrim(trim.getOrDefault(true))
+                            .withIncludeEntryDate(includeEntryDate)
+                            .withVersionDate(versionDate)
+                            .build();
+                    streamCsv(ctx, csvConfig, pageSize, dao, requestParameters);
+                    return;
+                }
+
                 if (versionDate != null) {
                     throw new IllegalArgumentException(String.format("Version date is only supported for:%s and %s",
                             Formats.JSONV2, Formats.XMLV2));
@@ -526,7 +583,6 @@ public void getAll(@NotNull Context ctx) {
                     format = "json";
                 }
 
-                String office = ctx.queryParam(OFFICE);
                 results = dao.getTimeseries(format, names, office, units, datum, beginZdt, endZdt, tz);
                 ctx.status(HttpServletResponse.SC_OK);
 
@@ -536,6 +592,7 @@ public void getAll(@NotNull Context ctx) {
                 ctx.header(Header.CONTENT_LENGTH, String.valueOf(bytes.length));
                 ctx.res.getOutputStream().write(bytes);
             }
+
             addDeprecatedContentTypeWarning(ctx, contentType);
         } catch (NotFoundException e) {
             CdaError re = new CdaError("Not found.");
@@ -554,6 +611,27 @@ public void getAll(@NotNull Context ctx) {
         }
     }
 
+    private void streamCsv(@NotNull Context ctx, CsvConfiguration csvConfig, int batchSize, TimeSeriesDao dao, TimeSeriesRequestParameters requestParameters) {
+        int csvBatchSize = validateCsvBatchSize(batchSize);
+        dao.streamRequestedTimeSeriesCsv(
+                requestParameters,
+                (stream, position, mediaType, totalLength) -> {
+                    ctx.status(HttpServletResponse.SC_OK);
+                    ctx.contentType(mediaType);
+                    ctx.header(Header.CONTENT_TYPE, Formats.CSV + "; charset=UTF-8");
+                    ctx.header("X-Stream-Batch-Size", String.valueOf(batchSize));
+                    try (stream) {
+                        IOUtils.copy(stream, ctx.res.getOutputStream());
+                    } catch (IOException e) {
+                        throw new RuntimeException(e);
+                    }
+                },
+                csvConfig,
+                null,
+                csvBatchSize //page-size drives streaming chunk size
+        );
+    }
+
     static RuntimeException unwrapExecutionException(java.util.concurrent.ExecutionException ex) {
         Throwable cause = ex.getCause();
         if (cause instanceof RuntimeException) {
@@ -566,6 +644,13 @@ static RuntimeException unwrapExecutionException(java.util.concurrent.ExecutionE
         return new RuntimeException(cause);
     }
 
+    private int validateCsvBatchSize(int requestedPageSize) {
+        if (requestedPageSize <= 0) {
+            throw new IllegalArgumentException("For CSV streaming, page-size must be greater than 0.");
+        }
+        return requestedPageSize;
+    }
+
     private void addLinkHeader(@NotNull Context ctx, TimeSeries ts, ContentType contentType) {
         try {
             // Send back the link to the next page in the response header
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dao/CsvOnDemandInputStream.java b/cwms-data-api/src/main/java/cwms/cda/data/dao/CsvOnDemandInputStream.java
new file mode 100644
index 0000000000..612022192f
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dao/CsvOnDemandInputStream.java
@@ -0,0 +1,178 @@
+package cwms.cda.data.dao;
+
+import com.google.common.flogger.FluentLogger;
+import cwms.cda.data.dto.csv.TimeSeriesCsv;
+import cwms.cda.data.dto.csv.TimeSeriesCsvRow;
+import cwms.cda.formatters.csv.CsvConfiguration;
+import cwms.cda.formatters.csv.CsvV1;
+import org.jooq.Cursor;
+import org.jooq.Record4;
+import org.jooq.exception.DataAccessException;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+// InputStream that renders CSV rows on-demand from a jOOQ Cursor
+final class CsvOnDemandInputStream extends InputStream {
+    private final Cursor<? extends Record4<Timestamp, Double, BigDecimal, Timestamp>> cursor;
+    private final Iterator<? extends Record4<Timestamp, Double, BigDecimal, Timestamp>> it;
+    private final CsvV1 csv;
+    private final String tsIdStr;
+    private final String officeId;
+    private final String units;
+    private final Timestamp versionTs;
+    private final int rowsPerBuffer;
+    private final CsvConfiguration csvConfiguration;
+    private final CsvConfiguration rowConfiguration;
+
+    private byte[] buffer = new byte[0];
+    private int bufPos = 0;
+    private boolean first = true;
+    private boolean closed = false;
+
+    CsvOnDemandInputStream(Cursor<? extends Record4<Timestamp, Double, BigDecimal, Timestamp>> cursor,
+                           CsvV1 csv,
+                           String tsIdStr,
+                           String officeId,
+                           String units,
+                           Timestamp versionTs,
+                           CsvConfiguration csvConfiguration,
+                           Integer rowsPerBuffer) {
+        this.cursor = cursor;
+        this.it = cursor.iterator();
+        this.csv = csv;
+        this.tsIdStr = tsIdStr;
+        this.officeId = officeId;
+        this.units = units;
+        this.versionTs = versionTs;
+        this.csvConfiguration = csvConfiguration;
+        this.rowConfiguration = new CsvConfiguration.Builder()
+                .from(csvConfiguration)
+                .withMetadataIncluded(false)
+                .build();
+
+        int rpb = rowsPerBuffer == null ? 1 : rowsPerBuffer;
+        this.rowsPerBuffer = rpb > 0 ? rpb : 1;
+    }
+
+    @Override
+    public int read() throws IOException {
+        byte[] one = new byte[1];
+        int r = read(one, 0, 1);
+        return r == -1 ? -1 : (one[0] & 0xFF);
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        if (closed) {
+            throw new IOException("Stream closed");
+        }
+        if (b == null) {
+            throw new NullPointerException("Buffer is null");
+        }
+        if (off < 0 || len < 0 || len > b.length - off) {
+            throw new IndexOutOfBoundsException();
+        }
+
+        int totalCopied = 0;
+        while (len > 0) {
+            if (bufPos >= buffer.length) {
+                if (!fillBuffer()) {
+                    break; // EOF
+                }
+            }
+
+            int toCopy = Math.min(len, buffer.length - bufPos);
+            System.arraycopy(buffer, bufPos, b, off, toCopy);
+            bufPos += toCopy;
+            off += toCopy;
+            len -= toCopy;
+            totalCopied += toCopy;
+        }
+
+        return totalCopied == 0 ? -1 : totalCopied;
+    }
+
+    private boolean fillBuffer() {
+        if (!it.hasNext()) {
+            buffer = new byte[0];
+            bufPos = 0;
+            return false;
+        }
+
+        List<TimeSeriesCsvRow> batch = new ArrayList<>(rowsPerBuffer);
+        int produced = 0;
+
+        while (it.hasNext() && produced < rowsPerBuffer) {
+            Record4<Timestamp, Double, BigDecimal, Timestamp> r = it.next();
+
+            Timestamp ts = r.value1();
+            Double val = r.value2();
+            BigDecimal qualityCode = r.value3();
+            Timestamp dataEntryDate = r.value4();
+
+            TimeSeriesCsvRow row = new TimeSeriesCsvRow.Builder()
+                    .withDateTime(ts == null ? null : ts.toInstant())
+                    .withValue(val)
+                    .withQualityCode(qualityCode == null ? null : qualityCode.intValue())
+                    .withDataEntryDate(dataEntryDate == null ? null : dataEntryDate.toInstant())
+                    .withUnits(units)
+                    .build();
+
+            batch.add(row);
+            produced++;
+        }
+
+        if (batch.isEmpty()) {
+            buffer = new byte[0];
+            bufPos = 0;
+            return false;
+        }
+
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withTimeSeriesId(tsIdStr)
+                .withOfficeId(officeId)
+                .withVersionDate(versionTs == null ? null : versionTs.toInstant().toString())
+                .withRows(batch)
+                .build();
+
+        String rendered = first
+                ? csv.format(container, csvConfiguration)
+                : csv.format(container, rowConfiguration);
+
+        if (first) {
+            first = false;
+        } else {
+            // Remove header from subsequent writes
+            int headerEnd = rendered.indexOf('\n');
+            // Check for \r\n as well
+            if (headerEnd != -1) {
+                if (headerEnd > 0 && rendered.charAt(headerEnd - 1) == '\r') {
+                    // It was \r\n, skip the \n and start at headerEnd + 1
+                }
+                rendered = rendered.substring(headerEnd + 1);
+            }
+        }
+        buffer = rendered.getBytes(StandardCharsets.UTF_8);
+        bufPos = 0;
+        return buffer.length > 0;
+    }
+
+    @Override
+    public void close() {
+        if (!closed) {
+            closed = true;
+            try {
+                cursor.close();
+            } catch (DataAccessException ex) {
+                FluentLogger.forEnclosingClass().atWarning().withCause(ex).log("Error closing database cursor");
+            }
+        }
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dao/JooqDao.java b/cwms-data-api/src/main/java/cwms/cda/data/dao/JooqDao.java
index 2c862083cd..742887fe6a 100644
--- a/cwms-data-api/src/main/java/cwms/cda/data/dao/JooqDao.java
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dao/JooqDao.java
@@ -835,7 +835,7 @@ private static UnsupportedOperationException buildUnsupportedOperationException(
         return new UnsupportedOperationException("CWMS currently does not support the requested operation", cause);
     }
 
-    private static @Nullable String sanitizeOrNull(@Nullable String localizedMessage) {
+    protected static @Nullable String sanitizeOrNull(@Nullable String localizedMessage) {
         if (localizedMessage != null && !localizedMessage.isEmpty()) {
             int length = localizedMessage.length();
             PolicyFactory sanitizer = new HtmlPolicyBuilder().disallowElements("<script>").toFactory();
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dao/StreamConsumer.java b/cwms-data-api/src/main/java/cwms/cda/data/dao/StreamConsumer.java
index 8e337ab413..6c59659ee0 100644
--- a/cwms-data-api/src/main/java/cwms/cda/data/dao/StreamConsumer.java
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dao/StreamConsumer.java
@@ -4,7 +4,20 @@
 import java.io.InputStream;
 import java.sql.SQLException;
 
+/**
+ * A consumer for streaming binary data out to callers (e.g., HTTP layer).
+ *
+ * The primary abstract method accepts an InputStream with optional position and total length.
+ * For generated content where the total length is not known up front (e.g., CSV rendered on-demand),
+ * use the default two-argument overload which delegates with null position/length values, allowing
+ * callers to choose chunked transfer without setting Content-Length.
+ */
 @FunctionalInterface
 public interface StreamConsumer {
+
     void accept(InputStream stream, long inputStreamPosition, String mediaType, long totalLength) throws SQLException, IOException;
+
+    default void accept(InputStream stream, String mediaType) throws SQLException, IOException {
+        accept(stream, 0L, mediaType, -1L);
+    }
 }
\ No newline at end of file
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDao.java b/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDao.java
index 424ebbc91e..bf4c7516fe 100644
--- a/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDao.java
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDao.java
@@ -5,6 +5,8 @@
 import cwms.cda.data.dto.RecentValue;
 import cwms.cda.data.dto.TimeSeries;
 import cwms.cda.data.dto.filteredtimeseries.FilteredTimeSeries;
+import cwms.cda.formatters.csv.CsvConfiguration;
+
 import java.sql.Timestamp;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -60,4 +62,10 @@ List<RecentValue> findRecentsInRange(String office, String categoryId, String gr
     List<RecentValue> findMostRecentsInRange(String office, List<String> tsIds, Timestamp pastLimit,
                                              Timestamp futureLimit, UnitSystem unitSystem);
 
+    void streamRequestedTimeSeriesCsv(TimeSeriesRequestParameters requestParameters,
+                                      StreamConsumer consumer,
+                                      CsvConfiguration csvConfig,
+                                      Integer dbFetchSize,
+                                      Integer rowsPerBuffer);
+
 }
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDaoImpl.java b/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDaoImpl.java
index d7ba536372..349fbf72ef 100644
--- a/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDaoImpl.java
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDaoImpl.java
@@ -1,6 +1,19 @@
 package cwms.cda.data.dao;
 
 import static com.google.common.flogger.LazyArgs.lazy;
+
+import cwms.cda.api.errors.InvalidItemException;
+import cwms.cda.data.dao.rsql.FieldResolver;
+import cwms.cda.data.dao.rsql.MapFieldResolver;
+import cwms.cda.data.dao.rsql.RSQLConditionBuilder;
+import cwms.cda.data.dto.filteredtimeseries.FilteredTimeSeries;
+import cwms.cda.data.dto.catalog.TimeSeriesAlias;
+import cwms.cda.formatters.csv.CsvConfiguration;
+import cwms.cda.helpers.DateUtils;
+
+import java.io.IOException;
+import java.sql.Connection;
+
 import static org.jooq.impl.DSL.asterisk;
 import static org.jooq.impl.DSL.countDistinct;
 import static org.jooq.impl.DSL.field;
@@ -10,6 +23,7 @@
 import static org.jooq.impl.DSL.partitionBy;
 import static org.jooq.impl.DSL.select;
 import static org.jooq.impl.DSL.selectDistinct;
+import static org.jooq.impl.DSL.selectOne;
 import static org.jooq.impl.DSL.table;
 import static usace.cwms.db.jooq.codegen.tables.AV_CWMS_TS_ID2.AV_CWMS_TS_ID2;
 import static usace.cwms.db.jooq.codegen.tables.AV_TS_EXTENTS_UTC.AV_TS_EXTENTS_UTC;
@@ -25,9 +39,6 @@
 import com.google.common.flogger.FluentLogger;
 import cwms.cda.api.enums.UnitSystem;
 import cwms.cda.api.enums.VersionType;
-import cwms.cda.data.dao.rsql.FieldResolver;
-import cwms.cda.data.dao.rsql.MapFieldResolver;
-import cwms.cda.data.dao.rsql.RSQLConditionBuilder;
 import cwms.cda.data.dto.Catalog;
 import cwms.cda.data.dto.CwmsDTOPaginated;
 import cwms.cda.data.dto.RecentValue;
@@ -38,15 +49,11 @@
 import cwms.cda.data.dto.TsvId;
 import cwms.cda.data.dto.VerticalDatumInfo;
 import cwms.cda.data.dto.catalog.CatalogEntry;
-import cwms.cda.data.dto.catalog.TimeSeriesAlias;
 import cwms.cda.data.dto.catalog.TimeseriesCatalogEntry;
-import cwms.cda.data.dto.filteredtimeseries.FilteredTimeSeries;
 import cwms.cda.formatters.xml.XMLv1;
-import cwms.cda.helpers.DateUtils;
 import cwms.cda.helpers.ZoneIdHelper;
 import java.math.BigDecimal;
 import java.math.BigInteger;
-import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.time.Duration;
@@ -99,6 +106,9 @@
 import usace.cwms.db.jooq.codegen.udt.records.ZTSV_ARRAY;
 import usace.cwms.db.jooq.codegen.udt.records.ZTSV_TYPE;
 
+import cwms.cda.formatters.csv.CsvV1;
+import cwms.cda.formatters.Formats;
+
 public class TimeSeriesDaoImpl extends JooqDao<TimeSeries> implements TimeSeriesDao {
     private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
@@ -220,6 +230,117 @@ public String getTimeseries(String format, String names, String office, String u
                 timezone.getId(), office);
     }
 
+    private ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> buildTsvDquQuery(
+            long tsCode, String officeId, String units,
+            TimeSeriesRequestParameters requestParameters,
+            boolean includeEntryDate) {
+        ZonedDateTime beginTime = requestParameters.getBeginTime();
+        ZonedDateTime endTime = requestParameters.getEndTime();
+        ZonedDateTime versionDate = requestParameters.getVersionDate();
+
+        Timestamp beginTimestamp = Timestamp.from(beginTime.toInstant());
+        Timestamp endTimestamp = Timestamp.from(endTime.toInstant());
+
+        AV_TSV_DQU view = AV_TSV_DQU.AV_TSV_DQU;
+
+        Field<BigDecimal> qualityForNormalization = DSL.nvl(
+                view.QUALITY_CODE.cast(BigDecimal.class),
+                DSL.val(BigDecimal.valueOf(5))
+        );
+
+        Field<BigDecimal> normalizedQuality = CWMS_TS_PACKAGE.call_NORMALIZE_QUALITY(
+                qualityForNormalization
+        ).as("quality_norm");
+
+        Field<Double> value = view.VALUE.as(VALUE);
+
+        Condition baseCondition = view.ALIASED_ITEM.isNull()
+                .and(view.TS_CODE.eq(tsCode))
+                .and(view.OFFICE_ID.eq(officeId))
+                .and(view.UNIT_ID.equalIgnoreCase(units))
+                .and(view.DATE_TIME.ge(beginTimestamp))
+                .and(view.DATE_TIME.le(endTimestamp))
+                .and(view.START_DATE.le(endTimestamp))
+                .and(view.END_DATE.gt(beginTimestamp));
+
+        ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> query;
+        if (versionDate != null) {
+            query = buildVersionedRowsQuery(
+                    view,
+                    value,
+                    normalizedQuality,
+                    baseCondition,
+                    versionDate,
+                    includeEntryDate
+            );
+        } else {
+            query = buildMaxVersionRowsQuery(
+                    view,
+                    value,
+                    normalizedQuality,
+                    baseCondition,
+                    includeEntryDate
+            );
+        }
+        return query;
+    }
+
+    @Override
+    public void streamRequestedTimeSeriesCsv(TimeSeriesRequestParameters requestParameters, StreamConsumer consumer,
+                                             CsvConfiguration csvConfiguration, Integer dbFetchSize, Integer rowsPerBuffer) {
+
+        boolean includeDataEntryDate = csvConfiguration.includeOptionalColumns();
+
+        DirectReadMetadata metadata = fetchRequestedTimeSeriesMetadataRecord(requestParameters);
+        if (metadata == null) {
+            throw new DataAccessException("Unable to resolve time series metadata for " + requestParameters.getNames());
+        }
+
+        long tsCode = metadata.tsCode;
+        String tsIdStr = metadata.tsId;
+        String officeResolved = metadata.officeId;
+        String resolvedUnits = metadata.units;
+
+        ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> query = buildTsvDquQuery(tsCode, officeResolved,
+                resolvedUnits, requestParameters, includeDataEntryDate);
+
+        logger.atFine().log("%s", lazy(query::getSQL));
+
+        int effectiveFetchSize = (dbFetchSize != null && dbFetchSize > 0)
+                ? dbFetchSize
+                : 1000;
+
+        if (rowsPerBuffer != null && rowsPerBuffer > 0 && effectiveFetchSize < rowsPerBuffer) {
+            effectiveFetchSize = rowsPerBuffer;
+        }
+
+        query.fetchSize(effectiveFetchSize);
+
+        ZonedDateTime versionDate = requestParameters.getVersionDate();
+        Timestamp versionTs = versionDate != null ? Timestamp.from(versionDate.toInstant()) : null;
+
+        try (Cursor<? extends Record4<Timestamp, Double, BigDecimal, Timestamp>> recCursor = query.fetchLazy()) {
+            CsvV1 csvFormatter = new CsvV1();
+
+            CsvOnDemandInputStream stream = new CsvOnDemandInputStream(
+                    recCursor,
+                    csvFormatter,
+                    tsIdStr,
+                    officeResolved,
+                    resolvedUnits,
+                    versionTs,
+                    csvConfiguration,
+                    rowsPerBuffer
+            );
+
+            try (stream) {
+                consumer.accept(stream, Formats.CSV);
+            } catch (IOException | SQLException e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
     /**
      * Retrieves a TimeSeries from the database
      * @param page an opaque token used for paging
@@ -727,7 +848,7 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,
 
         // Pagination happens after regular-interval gap rows are merged
         //  fetch the full raw window first
-        List<TimeSeries.Record> rawRows = fetchRequestedTimeSeriesRows(tsCode, metadataOfficeId, nativeUnits,
+        List<TimeSeries.Record> rawRows = fetchRequestedTimeSeriesRows(tsCode, metadataOfficeId,
                 metadataUnits, requestParameters, includeEntryDate);
         long effectiveIntervalOffset = intervalOffset;
         if (isRegularSeries(intervalMinutes, intervalPart)) {
@@ -774,6 +895,8 @@ private DirectReadMetadata fetchRequestedTimeSeriesMetadataRecord(
         final Field<String> tsId = CWMS_TS_PACKAGE.call_GET_TS_ID__2(DSL.val(names), officeId);
         final Field<BigDecimal> tsCode = CWMS_TS_PACKAGE.call_GET_TS_CODE__2(DSL.val(names), officeId);
 
+        validateUnits(units, tsCode, officeId, names);
+
         Table<Record3<BigDecimal, String, String>> validTs =
                 select(tsCode.as("tscode"),
                         tsId.as("tsid"),
@@ -839,42 +962,38 @@ private DirectReadMetadata fetchRequestedTimeSeriesMetadataRecord(
                 record.getValue("version_flag", String.class)));
     }
 
-    private List<TimeSeries.Record> fetchRequestedTimeSeriesRows(long tsCode, String officeId, String nativeUnits,
-                                                                 String requestedUnits,
-                                                                 TimeSeriesRequestParameters requestParameters,
-                                                                 boolean includeEntryDate) {
-        ZonedDateTime beginTime = requestParameters.getBeginTime();
-        ZonedDateTime endTime = requestParameters.getEndTime();
-        ZonedDateTime versionDate = requestParameters.getVersionDate();
-        Timestamp beginTimestamp = Timestamp.from(beginTime.toInstant());
-        Timestamp endTimestamp = Timestamp.from(endTime.toInstant());
+    private void validateUnits(String units, Field<BigDecimal> tsCode, Field<String> officeId, String name) {
+        if(!units.equalsIgnoreCase("SI") && !units.equalsIgnoreCase("EN")) {
 
-        AV_TSV_DQU view = AV_TSV_DQU.AV_TSV_DQU;
-        Field<BigDecimal> qualityForNormalization = DSL.nvl(
-                view.QUALITY_CODE.cast(BigDecimal.class),
-                DSL.val(BigDecimal.valueOf(5))
-        );
-        Field<BigDecimal> normalizedQuality = CWMS_TS_PACKAGE.call_NORMALIZE_QUALITY(
-                qualityForNormalization).as("quality_norm");
-        Field<Double> convertedValue = CWMS_UTIL_PACKAGE.call_CONVERT_UNITS(
-                view.VALUE, view.UNIT_ID, DSL.val(requestedUnits, String.class)).as(VALUE);
-
-        Condition baseCondition = view.ALIASED_ITEM.isNull()
-                .and(view.TS_CODE.eq(tsCode))
-                .and(view.OFFICE_ID.eq(officeId))
-                .and(view.UNIT_ID.equalIgnoreCase(nativeUnits))
-                .and(view.DATE_TIME.ge(beginTimestamp))
-                .and(view.DATE_TIME.le(endTimestamp))
-                .and(view.START_DATE.le(endTimestamp))
-                .and(view.END_DATE.gt(beginTimestamp));
+            boolean tsExists = dsl.fetchExists(
+                    selectOne()
+                            .from(AV_TSV_DQU.AV_TSV_DQU)
+                            .where(AV_TSV_DQU.AV_TSV_DQU.TS_CODE.eq(tsCode.cast(Long.class)))
+                            .and(AV_TSV_DQU.AV_TSV_DQU.OFFICE_ID.eq(officeId))
+            );
+            if(tsExists) {
+                boolean unitRequestedExists = dsl.fetchExists(
+                        selectOne()
+                                .from(AV_TSV_DQU.AV_TSV_DQU)
+                                .where(AV_TSV_DQU.AV_TSV_DQU.TS_CODE.eq(tsCode.cast(Long.class)))
+                                .and(AV_TSV_DQU.AV_TSV_DQU.OFFICE_ID.eq(officeId))
+                                .and(AV_TSV_DQU.AV_TSV_DQU.UNIT_ID.equalIgnoreCase(units))
+                );
 
-        ResultQuery<? extends Record4<Timestamp, Double, BigDecimal, Timestamp>> query;
-        if (versionDate != null) {
-            query = buildVersionedRowsQuery(view, convertedValue, normalizedQuality, baseCondition, versionDate,
-                    includeEntryDate);
-        } else {
-            query = buildMaxVersionRowsQuery(view, convertedValue, normalizedQuality, baseCondition, includeEntryDate);
+                if (!unitRequestedExists) {
+                    String msg = sanitizeOrNull(units + " is not a valid unit for time series " + name);
+                    throw new InvalidItemException(msg, new IllegalArgumentException(msg));
+                }
+            }
         }
+    }
+
+    private List<TimeSeries.Record> fetchRequestedTimeSeriesRows(long tsCode, String officeId,
+                                                                 String requestedUnits,
+                                                                 TimeSeriesRequestParameters requestParameters,
+                                                                 boolean includeEntryDate) {
+        ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> query = buildTsvDquQuery(tsCode, officeId,
+                requestedUnits, requestParameters, includeEntryDate);
 
         logger.atFine().log("%s", lazy(() -> query.getSQL(ParamType.INLINED)));
 
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvRequiredColumn.java b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvRequiredColumn.java
new file mode 100644
index 0000000000..d5a3189b54
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvRequiredColumn.java
@@ -0,0 +1,14 @@
+package cwms.cda.data.dto.csv;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Indicates that a CSV column is required, i.e. not optional
+ */
+@Target({ElementType.FIELD, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface CsvRequiredColumn {
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvUnitHeader.java b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvUnitHeader.java
new file mode 100644
index 0000000000..7da7b8e89c
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CsvUnitHeader.java
@@ -0,0 +1,22 @@
+package cwms.cda.data.dto.csv;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation to indicate that a field should have units included in its CSV header.
+ * For example, if a field "value" is annotated with @CsvUnitHeader, and the DTO
+ * has a "units" field with value "ft", the CSV header for this column will be
+ * "value (ft)".
+ */
+@Target({ElementType.FIELD, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface CsvUnitHeader {
+    /**
+     * The name(s) of the field(s) that should have units included in their CSV header.
+     * The units will be retrieved from the field/method annotated with this annotation.
+     */
+    String field();
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CwmsCsvDTO.java b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CwmsCsvDTO.java
new file mode 100644
index 0000000000..53b07f5e05
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/CwmsCsvDTO.java
@@ -0,0 +1,7 @@
+package cwms.cda.data.dto.csv;
+
+import java.util.List;
+
+public interface CwmsCsvDTO<T> {
+    List<T> getRows();
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsv.java b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsv.java
new file mode 100644
index 0000000000..ac52706145
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsv.java
@@ -0,0 +1,76 @@
+package cwms.cda.data.dto.csv;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import cwms.cda.data.dto.CwmsDTOBase;
+import cwms.cda.formatters.Formats;
+import cwms.cda.formatters.annotations.FormattableWith;
+import cwms.cda.formatters.csv.CsvRows;
+import cwms.cda.formatters.csv.CsvV1;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonNaming(PropertyNamingStrategies.KebabCaseStrategy.class)
+@JsonDeserialize(builder = TimeSeriesCsv.Builder.class)
+@FormattableWith(contentType = Formats.CSV, formatter = CsvV1.class, aliases = {Formats.DEFAULT})
+public final class TimeSeriesCsv extends CwmsDTOBase implements CwmsCsvDTO<TimeSeriesCsvRow> {
+
+    private final String timeSeriesId;
+    private final String officeId;
+    private final String versionDate;
+
+    @CsvRows
+    private final List<TimeSeriesCsvRow> rows;
+
+    private TimeSeriesCsv(Builder builder) {
+        this.timeSeriesId = builder.timeSeriesId;
+        this.officeId = builder.officeId;
+        this.versionDate = builder.versionDate;
+        this.rows = builder.rows;
+    }
+
+    public String getTimeSeriesId() { return timeSeriesId; }
+    public String getOfficeId() { return officeId; }
+    public String getVersionDate() { return versionDate; }
+
+    @Override
+    public List<TimeSeriesCsvRow> getRows() {
+        return rows;
+    }
+
+    @JsonNaming(PropertyNamingStrategies.KebabCaseStrategy.class)
+    public static final class Builder {
+        private String timeSeriesId;
+        private String officeId;
+        private String versionDate;
+        private List<TimeSeriesCsvRow> rows;
+
+        public Builder withTimeSeriesId(String timeSeriesId) {
+            this.timeSeriesId = timeSeriesId;
+            return this;
+        }
+
+        public Builder withOfficeId(String officeId) {
+            this.officeId = officeId;
+            return this;
+        }
+
+        public Builder withVersionDate(String versionDate) {
+            this.versionDate = versionDate;
+            return this;
+        }
+
+        public Builder withRows(List<TimeSeriesCsvRow> rows) {
+            this.rows = rows;
+            return this;
+        }
+
+        public TimeSeriesCsv build() {
+            return new TimeSeriesCsv(this);
+        }
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsvRow.java b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsvRow.java
new file mode 100644
index 0000000000..4665778741
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/data/dto/csv/TimeSeriesCsvRow.java
@@ -0,0 +1,91 @@
+package cwms.cda.data.dto.csv;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import cwms.cda.data.dto.CwmsDTOBase;
+
+import java.time.Instant;
+
+/**
+ * Single DTO for TimeSeries CSV rows. All potential columns exist on this class;
+ * only date-time and value are considered required and are annotated with @CsvRequiredColumn
+ * so that when using the CsvV1 mapper they are the only columns included. All other
+ * fields are considered optional and will only be included if the CsvConfiguration specified including them
+ */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonNaming(PropertyNamingStrategies.KebabCaseStrategy.class)
+@JsonDeserialize(builder = TimeSeriesCsvRow.Builder.class)
+public final class TimeSeriesCsvRow extends CwmsDTOBase {
+
+    @CsvRequiredColumn
+    @JsonProperty(index = 0)
+    private final Instant dateTime;
+
+    @CsvRequiredColumn
+    @JsonProperty(index = 1)
+    private final Double value;
+
+    @JsonProperty(index = 2)
+    private final Instant dataEntryDate;
+
+    @JsonProperty(index = 3)
+    private final Integer qualityCode;
+
+    @CsvUnitHeader(field = "value")
+    private final String units;
+
+    private TimeSeriesCsvRow(Builder builder) {
+        this.dateTime = builder.dateTime;
+        this.value = builder.value;
+        this.qualityCode = builder.qualityCode;
+        this.dataEntryDate = builder.dataEntryDate;
+        this.units = builder.units;
+    }
+
+    public Instant getDateTime() { return dateTime; }
+    public Double getValue() { return value; }
+    public Integer getQualityCode() { return qualityCode; }
+    public Instant getDataEntryDate() { return dataEntryDate; }
+    public String getUnits() { return units; }
+
+    @JsonNaming(PropertyNamingStrategies.KebabCaseStrategy.class)
+    public static final class Builder {
+        private Instant dateTime;
+        private Double value;
+        private Instant dataEntryDate;
+        private Integer qualityCode;
+        private String units;
+
+        public Builder withDateTime(Instant dateTime) {
+            this.dateTime = dateTime;
+            return this;
+        }
+
+        public Builder withValue(Double value) {
+            this.value = value;
+            return this;
+        }
+
+        public Builder withDataEntryDate(Instant dataEntryDate) {
+            this.dataEntryDate = dataEntryDate;
+            return this;
+        }
+
+        public Builder withQualityCode(Integer qualityCode) {
+            this.qualityCode = qualityCode;
+            return this;
+        }
+
+        public Builder withUnits(String units) {
+            this.units = units;
+            return this;
+        }
+
+        public TimeSeriesCsvRow build() {
+            return new TimeSeriesCsvRow(this);
+        }
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormat.java b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormat.java
new file mode 100644
index 0000000000..2a8bb3fd4d
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormat.java
@@ -0,0 +1,133 @@
+package cwms.cda.formatters;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Objects;
+
+public abstract class DateFormat {
+
+    private DateFormat() {
+
+    }
+
+    public abstract void apply(ObjectMapper mapper, JavaTimeModule module);
+
+    public static DateFormat epochMillis() {
+        return new EpochMillis();
+    }
+
+    public static DateFormat pattern(String pattern) {
+        return new PatternFormat(pattern);
+    }
+
+    public static final class EpochMillis extends DateFormat {
+        @Override
+        public void apply(ObjectMapper mapper, JavaTimeModule module) {
+            mapper.enable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            return obj instanceof EpochMillis;
+        }
+    }
+
+    public static final class PatternFormat extends DateFormat {
+        private final String pattern;
+
+        private PatternFormat(String pattern) {
+            this.pattern = pattern;
+        }
+
+        @Override
+        public void apply(ObjectMapper mapper, JavaTimeModule module) {
+            mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+            mapper.setDateFormat(new SimpleDateFormat(pattern));
+            setCustomJavaTimeFormat(pattern, module);
+        }
+
+        private static void setCustomJavaTimeFormat(String dateFormatPattern, JavaTimeModule javaTimeModule) {
+            DateTimeFormatter formatter =
+                    DateTimeFormatter.ofPattern(dateFormatPattern).withZone(ZoneOffset.UTC);
+
+            javaTimeModule.addSerializer(Instant.class, new JsonSerializer<>() {
+                @Override
+                public void serialize(Instant value, JsonGenerator gen, SerializerProvider serializers)
+                        throws IOException {
+                    if (value == null) {
+                        gen.writeNull();
+                    } else {
+                        gen.writeString(formatter.format(value));
+                    }
+                }
+            });
+
+            javaTimeModule.addDeserializer(Instant.class, new JsonDeserializer<>() {
+                @Override
+                public Instant deserialize(JsonParser p, DeserializationContext ctxt)
+                        throws IOException {
+                    String text = p.getText();
+                    if(text == null || text.isBlank())
+                    {
+                        return null;
+                    }
+                    return Instant.from(formatter.parse(text));
+                }
+            });
+
+            javaTimeModule.addSerializer(ZonedDateTime.class, new JsonSerializer<>() {
+                @Override
+                public void serialize(ZonedDateTime value, JsonGenerator gen, SerializerProvider serializers)
+                        throws IOException {
+                    if (value == null) {
+                        gen.writeNull();
+                    } else {
+                        gen.writeString(formatter.format(value.toInstant()));
+                    }
+                }
+            });
+            javaTimeModule.addDeserializer(ZonedDateTime.class, new JsonDeserializer<>() {
+                @Override
+                public ZonedDateTime deserialize(JsonParser p, DeserializationContext ctxt)
+                        throws IOException {
+                    String text = p.getText();
+                    if(text == null || text.isBlank())
+                    {
+                        return null;
+                    }
+                    return ZonedDateTime.ofInstant(
+                            Instant.from(formatter.parse(text)),
+                            ZoneOffset.UTC
+                    );
+                }
+            });
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (o == null || getClass() != o.getClass()) return false;
+            PatternFormat that = (PatternFormat) o;
+            return Objects.equals(pattern, that.pattern);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hashCode(pattern);
+        }
+    }
+
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatParameter.java b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatParameter.java
new file mode 100644
index 0000000000..ad9dac8fda
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatParameter.java
@@ -0,0 +1,29 @@
+package cwms.cda.formatters;
+
+public enum DateFormatParameter {
+    EPOCH_MILLIS("epoch-millis"),
+    ISO_INSTANT("ISO8601-Instant"),
+    ISO_OFFSET("ISO8601-Offset"),
+    ISO_LOCAL("ISO8601-Local"),
+    DATE_ONLY("date-only"),
+    CUSTOM("custom");
+
+    private final String value;
+
+    DateFormatParameter(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public static DateFormatParameter get(String value) {
+        for (DateFormatParameter format : DateFormatParameter.values()) {
+            if (format.value.equalsIgnoreCase(value)) {
+                return format;
+            }
+        }
+        return null;
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatResolver.java b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatResolver.java
new file mode 100644
index 0000000000..bf6e5e6ffb
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/DateFormatResolver.java
@@ -0,0 +1,43 @@
+package cwms.cda.formatters;
+
+
+public final class DateFormatResolver {
+    public static final String ISO_INSTANT_PATTERN = "yyyy-MM-dd'T'HH:mm:ss'Z'";
+    public static final String ISO_OFFSET_PATTERN = "yyyy-MM-dd'T'HH:mm:ssXXX";
+    public static final String ISO_LOCAL_PATTERN = "yyyy-MM-dd'T'HH:mm:ss";
+    public static final String DATE_ONLY_PATTERN = "yyyy-MM-dd";
+
+    private DateFormatResolver() {
+    }
+
+    public static DateFormat resolve(String dateFormatParam, String customPattern) {
+        if (dateFormatParam == null || dateFormatParam.isEmpty()) {
+            return DateFormat.pattern(ISO_INSTANT_PATTERN);
+        }
+
+        DateFormatParameter format = DateFormatParameter.get(dateFormatParam);
+        if (format == null) {
+            throw new UnsupportedOperationException("Unsupported date-format: " + dateFormatParam);
+        }
+
+        switch (format) {
+            case EPOCH_MILLIS:
+                return DateFormat.epochMillis();
+            case ISO_INSTANT:
+                return DateFormat.pattern(ISO_INSTANT_PATTERN);
+            case ISO_OFFSET:
+                return DateFormat.pattern(ISO_OFFSET_PATTERN);
+            case ISO_LOCAL:
+                return DateFormat.pattern(ISO_LOCAL_PATTERN);
+            case DATE_ONLY:
+                return DateFormat.pattern(DATE_ONLY_PATTERN);
+            case CUSTOM:
+                if (customPattern == null || customPattern.isEmpty()) {
+                    throw new IllegalArgumentException("date-format-pattern is required when date-format is set to 'custom'");
+                }
+                return DateFormat.pattern(customPattern);
+            default:
+                throw new UnsupportedOperationException("Unsupported date-format: " + dateFormatParam);
+        }
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/Formats.java b/cwms-data-api/src/main/java/cwms/cda/formatters/Formats.java
index 2ab7d01325..121911bac5 100644
--- a/cwms-data-api/src/main/java/cwms/cda/formatters/Formats.java
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/Formats.java
@@ -247,40 +247,32 @@ public static <T extends CwmsDTOBase> List<T> parseContentList(ContentType type,
     }
 
     /**
-     * Parses the supplied header param or queryParam to determine the content type.
-     * If both are supplied an exception is thrown.  If neither are supplied an exception is thrown.
+     * Parses the supplied header param and/or queryParam to determine the content type.
+     * Query parameter takes priority over the header and is parsed the same way as the header
+     * (i.e., supports full content types, versions, and DTO-specific aliases). If neither is
+     * supplied an exception is thrown.
      *
      * @param header     Accept header value
      * @param queryParam format query parameter value
      * @param klass      DTO object class, used for identifying content type aliases from the DTO's
      *                   <code>FormattableWith</code> annotations.
      * @return an appropriate standard mimetype for lookup
-     * @throws FormattingException if the header and queryParam are both supplied or neither are
+     * @throws FormattingException if neither header nor queryParam can be parsed into a supported content type
      */
-    public static ContentType parseHeaderAndQueryParm(String header, String queryParam,
-        Class<? extends CwmsDTOBase> klass) {
+    public static ContentType parseHeaderAndQueryParm(String header, String queryParam, Class<? extends CwmsDTOBase> klass) {
+        // If a query parameter is provided, it overrides the header.
         if (queryParam != null && !queryParam.isEmpty()) {
-            if (header != null && !header.isEmpty() && !DEFAULT.equals(header.trim())) {
-                // If the user supplies an accept header and also a format= parameter, which
-                // should we use?
-                // The older format= query parameters don't give us the option to supply a
-                // version the
-                // way that the accept header does.
-                throw new UnsupportedFormatException("Accept header and query parameter are both "
-                        + "present, this is not supported.");
-            }
-
             ContentType ct = parseQueryParam(queryParam, klass);
             if (ct != null) {
                 return ct;
-            } else {
-                throw new UnsupportedFormatException("content-type " + queryParam + " is not implemented");
             }
-        } else if (header == null) {
+        }
+
+        // No query parameter provided; use the header (parseHeader handles null/empty by mapping to */*)
+        if (header == null) {
             throw new UnsupportedFormatException("no content type or format specified");
-        } else {
-            return parseHeader(header, klass);
         }
+        return parseHeader(header, klass);
     }
 
     /**
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvConfiguration.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvConfiguration.java
new file mode 100644
index 0000000000..901f8be8b3
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvConfiguration.java
@@ -0,0 +1,60 @@
+package cwms.cda.formatters.csv;
+
+import cwms.cda.formatters.DateFormatResolver;
+import cwms.cda.formatters.DateFormat;
+
+public final class CsvConfiguration {
+    private final boolean includeMetadata;
+    private final boolean includeOptionalColumns;
+    private final DateFormat dateFormat;
+
+    private CsvConfiguration(Builder builder) {
+        this.includeMetadata = builder.includeMetadata;
+        this.includeOptionalColumns = builder.includeOptionalColumns;
+        this.dateFormat = builder.dateFormat;
+    }
+
+    public boolean includeMetadata() {
+        return includeMetadata;
+    }
+
+    public boolean includeOptionalColumns() {
+        return includeOptionalColumns;
+    }
+
+    public DateFormat getDateFormat() {
+        return dateFormat;
+    }
+
+    public static class Builder {
+        private boolean includeMetadata = false;
+        private boolean includeOptionalColumns = false;
+        private DateFormat dateFormat = DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN);
+
+        public Builder withMetadataIncluded(boolean includeMetadata) {
+            this.includeMetadata = includeMetadata;
+            return this;
+        }
+
+        public Builder withOptionalColumnsIncluded(boolean includeOptionalColumns) {
+            this.includeOptionalColumns = includeOptionalColumns;
+            return this;
+        }
+
+        public Builder withDateFormat(DateFormat dateFormat) {
+            this.dateFormat = dateFormat;
+            return this;
+        }
+
+        public Builder from(CsvConfiguration config) {
+            this.includeMetadata = config.includeMetadata;
+            this.includeOptionalColumns = config.includeOptionalColumns;
+            this.dateFormat = config.dateFormat;
+            return this;
+        }
+
+        public CsvConfiguration build() {
+            return new CsvConfiguration(this);
+        }
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvExampleGenerator.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvExampleGenerator.java
new file mode 100644
index 0000000000..f072d1b9a5
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvExampleGenerator.java
@@ -0,0 +1,143 @@
+package cwms.cda.formatters.csv;
+
+import com.google.common.flogger.FluentLogger;
+import cwms.cda.data.dto.csv.CwmsCsvDTO;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+public class CsvExampleGenerator {
+
+    public static String getExample(Class<? extends CwmsCsvDTO<?>> dtoClass) {
+        try {
+            CwmsCsvDTO<?> dto = createDummyInstance(dtoClass);
+            //show everything we can in the example, so include metadata and optional columns
+            return new CsvV1().format(dto, new CsvConfiguration.Builder()
+                    .withMetadataIncluded(true)
+                    .withOptionalColumnsIncluded(true)
+                    .build());
+        } catch (Exception e) {
+            //don't want to stop the world if we can't generate an example, but log the error for debugging
+            FluentLogger.forEnclosingClass().atInfo().withCause(e).log("Failed to create csv example for " + dtoClass.getName() + ": ", e);
+        }
+        return "Could not generate example for " + dtoClass.getName();
+    }
+
+    private static <T extends CwmsCsvDTO<?>> T createDummyInstance(Class<T> dtoClass) throws Exception {
+        Class<?> builderClass = null;
+        for (Class<?> inner : dtoClass.getDeclaredClasses()) {
+            if (inner.getSimpleName().equals("Builder")) {
+                builderClass = inner;
+                break;
+            }
+        }
+
+        if (builderClass == null) {
+            Constructor<T> constr = dtoClass.getDeclaredConstructor();
+            constr.setAccessible(true);
+            return constr.newInstance();
+        }
+
+        Constructor<?> builderConstr = builderClass.getDeclaredConstructor();
+        builderConstr.setAccessible(true);
+        Object builder = builderConstr.newInstance();
+        
+        for (Field f : dtoClass.getDeclaredFields()) {
+            Object val = getDummyValue(f);
+            if (val != null) {
+                String methodName = "with" + f.getName().substring(0, 1).toUpperCase() + f.getName().substring(1);
+                try {
+                    Method m = builderClass.getMethod(methodName, f.getType());
+                    m.setAccessible(true);
+                    m.invoke(builder, val);
+                } catch (NoSuchMethodException ignored) {
+                    try {
+                        Method m2 = builderClass.getMethod(f.getName(), f.getType());
+                        m2.setAccessible(true);
+                        m2.invoke(builder, val);
+                    } catch (NoSuchMethodException ignored2) {
+                    }
+                }
+            }
+        }
+
+        Method buildMethod = builderClass.getMethod("build");
+        buildMethod.setAccessible(true);
+        return dtoClass.cast(buildMethod.invoke(builder));
+    }
+
+    private static Object getDummyValue(Field f) throws Exception {
+        Class<?> type = f.getType();
+        if (f.isAnnotationPresent(CsvRows.class) && List.class.isAssignableFrom(type)) {
+            ParameterizedType pt = (ParameterizedType) f.getGenericType();
+            Class<?> rowType = (Class<?>) pt.getActualTypeArguments()[0];
+            List<Object> rows = new ArrayList<>();
+            rows.add(createDummyRow(rowType));
+            return rows;
+        }
+
+        return getDummyValueSimple(type);
+    }
+
+    private static Object createDummyRow(Class<?> rowType) throws Exception {
+        Class<?> builderClass = null;
+        for (Class<?> inner : rowType.getDeclaredClasses()) {
+            if (inner.getSimpleName().equals("Builder")) {
+                builderClass = inner;
+                break;
+            }
+        }
+
+        if (builderClass == null) {
+            try {
+                Constructor<?> constr = rowType.getDeclaredConstructor();
+                constr.setAccessible(true);
+                return constr.newInstance();
+            } catch (NoSuchMethodException e) {
+                // If no default constructor and no builder, we might be in trouble for a generic generator.
+                // But most DTOs should have one or the other.
+                throw e;
+            }
+        }
+
+        Constructor<?> builderConstr = builderClass.getDeclaredConstructor();
+        builderConstr.setAccessible(true);
+        Object builder = builderConstr.newInstance();
+        for (Field f : rowType.getDeclaredFields()) {
+            Object val = getDummyValueSimple(f.getType());
+            if (val != null) {
+                String methodName = "with" + f.getName().substring(0, 1).toUpperCase() + f.getName().substring(1);
+                try {
+                    Method m = builderClass.getMethod(methodName, f.getType());
+                    m.setAccessible(true);
+                    m.invoke(builder, val);
+                } catch (NoSuchMethodException ignored) {
+                    // Try without "with" prefix as some builders use that
+                    try {
+                        Method m2 = builderClass.getMethod(f.getName(), f.getType());
+                        m2.setAccessible(true);
+                        m2.invoke(builder, val);
+                    } catch (NoSuchMethodException ignored2) {
+                    }
+                }
+            }
+        }
+        Method buildMethod = builderClass.getMethod("build");
+        buildMethod.setAccessible(true);
+        return buildMethod.invoke(builder);
+    }
+
+    private static Object getDummyValueSimple(Class<?> type) {
+        if (type == String.class) return "string";
+        if (type == Instant.class) return Instant.now();
+        if (type == Integer.class || type == int.class) return 0;
+        if (type == Double.class || type == double.class) return 0.0;
+        if (type == Long.class || type == long.class) return 0L;
+        if (type == Boolean.class || type == boolean.class) return false;
+        return null;
+    }
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvFormatter.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvFormatter.java
new file mode 100644
index 0000000000..5a6729663a
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvFormatter.java
@@ -0,0 +1,8 @@
+package cwms.cda.formatters.csv;
+
+import cwms.cda.data.dto.csv.CwmsCsvDTO;
+import cwms.cda.formatters.OutputFormatter;
+
+public interface CsvFormatter extends OutputFormatter {
+    String format(CwmsCsvDTO<?> dto, CsvConfiguration config);
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvRows.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvRows.java
new file mode 100644
index 0000000000..de89fe81c9
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvRows.java
@@ -0,0 +1,14 @@
+package cwms.cda.formatters.csv;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks a field as containing the data rows for a CSV DTO.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD, ElementType.METHOD})
+public @interface CsvRows {
+}
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvV1.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvV1.java
index 5f270fa851..06d556679c 100644
--- a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvV1.java
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CsvV1.java
@@ -1,56 +1,77 @@
 package cwms.cda.formatters.csv;
 
-import java.io.InputStream;
-import java.util.List;
-
+import com.fasterxml.jackson.core.JsonProcessingException;
 import cwms.cda.data.dto.CwmsDTOBase;
 import cwms.cda.data.dto.LocationGroup;
 import cwms.cda.data.dto.Office;
+import cwms.cda.data.dto.csv.CwmsCsvDTO;
 import cwms.cda.formatters.Formats;
-import cwms.cda.formatters.OutputFormatter;
+import cwms.cda.formatters.FormattingException;
+import java.io.InputStream;
+import java.util.List;
 
-public class CsvV1 implements OutputFormatter {
+public class CsvV1 implements CsvFormatter {
 
     @Override
     public String getContentType() {
         return Formats.CSV;
     }
 
+    /**
+     * Default formatting does not include metadata in either columns or comments.
+    **/
     @Override
     public String format(CwmsDTOBase dto) {
-        String retVal = null;
-        if (dto instanceof Office ) {
-            retVal = new CsvV1Office().format(dto);
-        } else if (dto instanceof LocationGroup ) {
-            retVal = new CsvV1LocationGroup().format(dto);
+        try {
+            if (dto instanceof Office ) {
+                return new CsvV1Office().format(dto);
+            } else if (dto instanceof LocationGroup ) {
+                return new CsvV1LocationGroup().format(dto);
+            } else if (dto instanceof CwmsCsvDTO) {
+                return format((CwmsCsvDTO<?>) dto, new CsvConfiguration.Builder().build());
+            } else {
+                throw new FormattingException(dto.getClass().getName() + " is not currently supported for CSV formatting.");
+            }
+        } catch (Exception e) {
+            throw new FormattingException("Could not serialize:" + dto.getClass().getName(), e);
         }
-        return retVal;
     }
 
+    @Override
+    public String format(CwmsCsvDTO<?> dto, CsvConfiguration config) {
+        try {
+            return CwmsCsvProcessor.formatCwmsCsv(dto, config);
+        } catch (JsonProcessingException e) {
+            throw new FormattingException("Could not serialize:" + dto.getClass().getName(), e);
+        }
+    }
+
+
     @Override
     public String format(List<? extends CwmsDTOBase> dtoList) {
-        String retVal = null;
         if (dtoList != null && !dtoList.isEmpty()) {
             CwmsDTOBase dto = dtoList.get(0);
             if (dto instanceof Office) {
-                retVal = new CsvV1Office().format(dtoList);
-            } else if(dto instanceof LocationGroup) {
-                retVal = new CsvV1LocationGroup().format(dtoList);
+                return new CsvV1Office().format(dtoList);
+            } else if (dto instanceof LocationGroup) {
+                return new CsvV1LocationGroup().format(dtoList);
+            } else {
+                throw new FormattingException(dto.getClass().getName() + " is not currently supported for CSV formatting.");
             }
-
         }
-        return retVal;
+        return null;
     }
 
     @Override
     public <T extends CwmsDTOBase> T parseContent(String content, Class<T> type) {
-        T retVal = null;
         if (type.isAssignableFrom(Office.class)) {
-            retVal = new CsvV1Office().parseContent(content, type);
+            return new CsvV1Office().parseContent(content, type);
         } else if (type.isAssignableFrom(LocationGroup.class)) {
-            retVal = new CsvV1LocationGroup().parseContent(content, type);
+            return new CsvV1LocationGroup().parseContent(content, type);
+        } else if (CwmsCsvDTO.class.isAssignableFrom(type)) {
+            return CwmsCsvProcessor.parseCwmsCsv(content, type);
         }
-        return retVal;
+        return null;
     }
 
     @Override
@@ -63,4 +84,5 @@ public <T extends CwmsDTOBase> T parseContent(InputStream content, Class<T> type
         }
         return retVal;
     }
+
 }
diff --git a/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CwmsCsvProcessor.java b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CwmsCsvProcessor.java
new file mode 100644
index 0000000000..967d8488e6
--- /dev/null
+++ b/cwms-data-api/src/main/java/cwms/cda/formatters/csv/CwmsCsvProcessor.java
@@ -0,0 +1,456 @@
+package cwms.cda.formatters.csv;
+
+import com.fasterxml.jackson.annotation.JsonFilter;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.ser.FilterProvider;
+import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
+import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
+import com.fasterxml.jackson.dataformat.csv.CsvGenerator;
+import com.fasterxml.jackson.dataformat.csv.CsvMapper;
+import com.fasterxml.jackson.dataformat.csv.CsvParser;
+import com.fasterxml.jackson.dataformat.csv.CsvSchema;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import cwms.cda.data.dto.CwmsDTOBase;
+import cwms.cda.data.dto.csv.CsvRequiredColumn;
+import cwms.cda.data.dto.csv.CsvUnitHeader;
+import cwms.cda.data.dto.csv.CwmsCsvDTO;
+import cwms.cda.formatters.FormattingException;
+import cwms.cda.formatters.DateFormat;
+import cwms.cda.formatters.json.adapters.ZoneIdDeserializer;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Utility class to processing cwms CSV DTOs for both reading and writing.
+ * Handles building header lines with units, building metadata comment lines, and parsing metadata and units from CSV content.
+ */
+public class CwmsCsvProcessor {
+
+    private static String buildMetadataComments(Object dto) {
+        StringBuilder sb = new StringBuilder();
+        Class<?> cls = dto.getClass();
+        List<Field> fields = getAllFields(cls);
+        for (Field f : fields) {
+            if (f.isAnnotationPresent(CsvRows.class)) {
+                continue;
+            }
+            f.setAccessible(true);
+            try {
+                Object val = f.get(dto);
+                if (val != null) {
+                    String key = resolveKeyName(f, cls);
+                    sb.append("# ").append(key).append(": ").append(val).append("\n");
+                }
+            } catch (IllegalAccessException ex) {
+                throw new FormattingException("Error building metadata comments for " + cls.getName(), ex);
+            }
+        }
+        return sb.toString();
+    }
+
+    private static List<Field> getAllFields(Class<?> cls) {
+        List<Field> fields = new ArrayList<>();
+        Class<?> current = cls;
+        while (current != null && current != Object.class) {
+            fields.addAll(Arrays.asList(current.getDeclaredFields()));
+            current = current.getSuperclass();
+        }
+        return fields;
+    }
+
+    private static String buildHeader(Object dto, boolean includeOptionalColumns) {
+        StringBuilder sb = new StringBuilder();
+        Map<String, String> fieldToUnits = new HashMap<>();
+
+        // Find row type and examples to get units from
+        Class<?> rowType = null;
+        Object firstRow = null;
+        if (dto instanceof CwmsCsvDTO) {
+            List<?> rows = ((CwmsCsvDTO<?>) dto).getRows();
+            if (rows != null && !rows.isEmpty()) {
+                firstRow = rows.get(0);
+                rowType = firstRow.getClass();
+            }
+        } else if (dto instanceof List && !((List<?>) dto).isEmpty()) {
+            firstRow = ((List<?>) dto).get(0);
+            rowType = firstRow.getClass();
+        } else {
+            rowType = dto.getClass();
+        }
+
+        try {
+            // Check top-level DTO first
+            extractFieldToUnits(dto, fieldToUnits);
+
+            // If it's a list or CwmsCsv, also check the row object for @CsvUnitHeader
+            if (firstRow != null) {
+                extractFieldToUnits(firstRow, fieldToUnits);
+            }
+        } catch (InvocationTargetException | IllegalAccessException e) {
+            throw new FormattingException("Error extracting units for CSV header", e);
+        }
+
+        if (rowType != null) {
+            List<ColumnInfo> columns = new ArrayList<>();
+            for (Field f : getAllFields(rowType)) {
+                JsonProperty jp = f.getAnnotation(JsonProperty.class);
+                if (jp != null && jp.index() != JsonProperty.INDEX_UNKNOWN) {
+                    if (!includeOptionalColumns && !f.isAnnotationPresent(CsvRequiredColumn.class)) {
+                        continue;
+                    }
+                    String name = resolvePropertyName(f);
+                    String units = fieldToUnits.get(name);
+                    if (units != null) {
+                        name = name + " (" + units + ")";
+                    }
+                    columns.add(new ColumnInfo(name, jp.index()));
+                }
+            }
+            columns.sort(Comparator.comparingInt(c -> c.order));
+            for (int i = 0; i < columns.size(); i++) {
+                sb.append(columns.get(i).name);
+                if (i < columns.size() - 1) {
+                    sb.append(",");
+                }
+            }
+            sb.append("\n");
+        }
+
+        return sb.toString();
+    }
+
+    private static void extractFieldToUnits(Object dto, Map<String, String> fieldToUnits) throws IllegalAccessException, InvocationTargetException {
+        if (dto == null) return;
+        Class<?> cls = dto.getClass();
+        List<Field> fields = getAllFields(cls);
+        for (Field f : fields) {
+            CsvUnitHeader ann = f.getAnnotation(CsvUnitHeader.class);
+            if (ann != null) {
+                f.setAccessible(true);
+                Object val = f.get(dto);
+                if (val != null) {
+                    String unitVal = val.toString();
+                    if (!unitVal.isEmpty()) {
+                        fieldToUnits.put(ann.field(), unitVal);
+                    }
+                }
+            }
+        }
+        List<Method> methods = getAllMethods(cls);
+        for (Method m : methods) {
+            CsvUnitHeader ann = m.getAnnotation(CsvUnitHeader.class);
+            if (ann != null && m.getParameterCount() == 0) {
+                m.setAccessible(true);
+                Object val = m.invoke(dto);
+                if (val != null) {
+                    String unitVal = val.toString();
+                    if (!unitVal.isEmpty()) {
+                        fieldToUnits.put(ann.field(), unitVal);
+                    }
+                }
+            }
+        }
+    }
+
+    static <T extends CwmsDTOBase> T parseCwmsCsv(String content, Class<T> type) {
+        try {
+            T dto;
+            try {
+                dto = type.getDeclaredConstructor().newInstance();
+            } catch (NoSuchMethodException e) {
+                // No default constructor, try to find a Builder
+                Class<?> builderClass = null;
+                for (Class<?> inner : type.getDeclaredClasses()) {
+                    if (inner.getSimpleName().equals("Builder")) {
+                        builderClass = inner;
+                        break;
+                    }
+                }
+                if (builderClass != null) {
+                    Object builder = builderClass.getDeclaredConstructor().newInstance();
+                    Method buildMethod = builderClass.getDeclaredMethod("build");
+                    dto = type.cast(buildMethod.invoke(builder));
+                } else {
+                    throw e;
+                }
+            }
+
+            if (dto instanceof CwmsCsvDTO) {
+                Map<String, String> metadata = CwmsCsvProcessor.parseMetadata(content);
+                String units = CwmsCsvProcessor.parseUnits(content);
+                CwmsCsvProcessor.injectMetadataAndUnits(content, type, dto, metadata, units);
+                return dto;
+            }
+        } catch (Exception e) {
+            throw new FormattingException("Could not parse " + type.getName(), e);
+        }
+        throw new FormattingException("Could not parse " + type.getName() + ". Must be a " + CwmsCsvDTO.class.getName());
+    }
+
+    private static <T extends CwmsDTOBase> void injectMetadataAndUnits(String content, Class<T> type, T dto, Map<String, String> metadata, String units) throws IOException, IllegalAccessException {
+        // Inject metadata into DTO
+        CwmsCsvProcessor.applyMetadataAndUnits(dto, metadata, units);
+
+        Field rowsField = getRowsField(type);
+
+        if (rowsField != null) {
+            rowsField.setAccessible(true);
+            Class<?> rowType = (Class<?>) ((ParameterizedType) rowsField.getGenericType()).getActualTypeArguments()[0];
+            List<?> rows = parseRows(content, rowType);
+
+            if (units != null) {
+                for (Object row : rows) {
+                    CwmsCsvProcessor.applyMetadataAndUnits(row, metadata, units);
+                }
+            }
+
+            rowsField.set(dto, rows);
+        }
+    }
+
+    private static List<?> parseRows(String content, Class<?> csvRowDtoType) throws IOException {
+        CsvMapper csvMapper = buildObjectMapper(csvRowDtoType, new CsvConfiguration.Builder().build());
+        csvMapper.enable(CsvParser.Feature.ALLOW_COMMENTS);
+        csvMapper.enable(CsvParser.Feature.SKIP_EMPTY_LINES);
+        csvMapper.enable(CsvParser.Feature.TRIM_SPACES);
+        CsvSchema schema = csvMapper.schemaFor(csvRowDtoType).withHeader();
+        try (MappingIterator<?> it = csvMapper.readerFor(csvRowDtoType).with(schema).readValues(content)) {
+            return it.readAll();
+        }
+    }
+
+    static String formatCwmsCsv(CwmsCsvDTO<?> dto, CsvConfiguration config) throws JsonProcessingException {
+        StringBuilder sb = new StringBuilder();
+
+        if (config.includeMetadata()) {
+            sb.append(buildMetadataComments(dto));
+        }
+
+        List<?> rows = dto.getRows();
+        if (rows != null && !rows.isEmpty()) {
+            Object firstRow = rows.get(0);
+
+            CsvMapper csvMapper = buildObjectMapper(firstRow.getClass(), config);
+
+            CsvSchema.Builder schemaBuilder = CsvSchema.builder();
+
+            List<Field> fields = getAllFields(firstRow.getClass());
+
+            fields.stream()
+                    .filter(f -> {
+                        JsonProperty jp = f.getAnnotation(JsonProperty.class);
+                        if (jp == null || jp.index() == JsonProperty.INDEX_UNKNOWN) {
+                            return false;
+                        }
+                        return config.includeOptionalColumns() || f.isAnnotationPresent(CsvRequiredColumn.class);
+                    })
+                    .sorted(Comparator.comparingInt(f -> f.getAnnotation(JsonProperty.class).index()))
+                    .forEach(f -> schemaBuilder.addColumn(resolvePropertyName(f)));
+
+            CsvSchema schema = schemaBuilder.build();
+            String header = buildHeader(dto, config.includeOptionalColumns());
+            sb.append(header);
+            FilterProvider filters = new SimpleFilterProvider()
+                    .addFilter("columnFilter", SimpleBeanPropertyFilter.filterOutAllExcept(new HashSet<>(schema.getColumnNames())));
+            sb.append(csvMapper.writer(schema).with(filters).writeValueAsString(rows));
+        }
+        return sb.toString();
+    }
+
+    private static CsvMapper buildObjectMapper(Class<?> rowType, CsvConfiguration config) {
+        CsvMapper mapper = new CsvMapper();
+        // Without these two disables an Instant gets written as 3333333.335000000
+        mapper.disable(SerializationFeature.WRITE_DATE_TIMESTAMPS_AS_NANOSECONDS);
+        mapper.disable(DeserializationFeature.READ_DATE_TIMESTAMPS_AS_NANOSECONDS);
+        mapper.disable(CsvGenerator.Feature.ALWAYS_QUOTE_STRINGS);
+        mapper.enable(CsvGenerator.Feature.STRICT_CHECK_FOR_QUOTING);
+        mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+
+        mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        JavaTimeModule javaTimeModule = new JavaTimeModule();
+        mapper.registerModule(new Jdk8Module());
+        mapper.enable(CsvParser.Feature.ALLOW_COMMENTS);
+        mapper.addMixIn(rowType, PropertyFilterMixIn.class);
+
+        DateFormat dateFormat = config.getDateFormat();
+        dateFormat.apply(mapper, javaTimeModule);
+        mapper.registerModule(javaTimeModule);
+
+        SimpleModule module = new SimpleModule();
+        module.addDeserializer(ZoneId.class, new ZoneIdDeserializer());
+        mapper.registerModule(module);
+        return mapper;
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonFilter("columnFilter")
+    abstract static class PropertyFilterMixIn {}
+
+    private static <T extends CwmsDTOBase> @Nullable Field getRowsField(Class<T> type) {
+        // Find rows field
+        Field rowsField = null;
+        for (Field f : CwmsCsvProcessor.getAllFields(type)) {
+            if (f.isAnnotationPresent(CsvRows.class)) {
+                rowsField = f;
+                break;
+            }
+        }
+        return rowsField;
+    }
+
+    private static List<Method> getAllMethods(Class<?> cls) {
+        List<Method> methods = new ArrayList<>();
+        Class<?> current = cls;
+        while (current != null && current != Object.class) {
+            methods.addAll(Arrays.asList(current.getDeclaredMethods()));
+            current = current.getSuperclass();
+        }
+        return methods;
+    }
+
+    private static String resolvePropertyName(Field f) {
+        JsonProperty jp = f.getAnnotation(JsonProperty.class);
+        if (jp != null && !jp.value().isEmpty()) {
+            return jp.value();
+        }
+        JsonNaming naming = f.getDeclaringClass().getAnnotation(JsonNaming.class);
+        return getName(f, naming);
+    }
+
+    private static String getName(Field f, JsonNaming naming) {
+        if (naming != null) {
+            try {
+                Object strat = naming.value().getDeclaredConstructor().newInstance();
+                if (strat instanceof PropertyNamingStrategies.NamingBase) {
+                    return ((PropertyNamingStrategies.NamingBase) strat).translate(f.getName());
+                }
+            } catch (InvocationTargetException | InstantiationException | IllegalAccessException | NoSuchMethodException e) {
+                throw new FormattingException("Error resolving property name for " + f.getName(), e);
+            }
+        }
+        return f.getName();
+    }
+
+    private static String resolveKeyName(Field f, Class<?> owner) {
+        JsonProperty jp = f.getAnnotation(JsonProperty.class);
+        if (jp != null && !jp.value().isEmpty()) {
+            return jp.value();
+        }
+        JsonNaming naming = owner.getAnnotation(JsonNaming.class);
+        return getName(f, naming);
+    }
+
+    private static Object convertToType(String val, Class<?> type) {
+        if (type == String.class) return val;
+        if (type == Instant.class) return Instant.parse(val);
+        if (type == Integer.class || type == int.class) return Integer.parseInt(val);
+        if (type == Double.class || type == double.class) return Double.parseDouble(val);
+        if (type == Long.class || type == long.class) return Long.parseLong(val);
+        if (type == Boolean.class || type == boolean.class) return Boolean.parseBoolean(val);
+        return null;
+    }
+
+    private static void applyMetadataAndUnits(Object dto, Map<String, String> metadata, String units) {
+        if (dto == null) return;
+        Class<?> cls = dto.getClass();
+        // Use a list of all fields including superclasses
+        List<Field> allFields = CwmsCsvProcessor.getAllFields(cls);
+
+        for (Field f : allFields) {
+            f.setAccessible(true);
+            try {
+                // Handle Units via @CsvUnitHeader
+                if (units != null) {
+                    CsvUnitHeader unitAnn = f.getAnnotation(CsvUnitHeader.class);
+                    if (unitAnn != null) {
+                        f.set(dto, units);
+                    }
+                }
+
+                // Handle Metadata
+                String key = CwmsCsvProcessor.resolveKeyName(f, cls);
+                String val = metadata.get(key);
+                if (val != null) {
+                    Object converted = CwmsCsvProcessor.convertToType(val, f.getType());
+                    if (converted != null) {
+                        f.set(dto, converted);
+                    }
+                }
+            } catch (IllegalAccessException e) {
+                throw new FormattingException("Error applying metadata to field " + f.getName(), e);
+            }
+        }
+    }
+
+    private static Map<String, String> parseMetadata(String content) {
+        Map<String, String> metadata = new HashMap<>();
+        String[] lines = content.split("\\r?\\n");
+        for (String line : lines) {
+            String trimmed = line.trim();
+            if (trimmed.startsWith("#")) {
+                String comment = trimmed.substring(1).trim();
+                int colon = comment.indexOf(':');
+                if (colon != -1) {
+                    String key = comment.substring(0, colon).trim();
+                    String val = comment.substring(colon + 1).trim();
+                    metadata.put(key, val);
+                }
+            } else if (!trimmed.isEmpty()) {
+                break; // Header or data starts
+            }
+        }
+        return metadata;
+    }
+
+    private static String parseUnits(String content) {
+        String[] lines = content.split("\\r?\\n");
+        for (String line : lines) {
+            String trimmed = line.trim();
+            if (trimmed.startsWith("#")) continue;
+            if (trimmed.isEmpty()) continue;
+            // First non-comment line is header
+            int start = trimmed.indexOf('(');
+            int end = trimmed.indexOf(')');
+            if (start != -1 && end != -1 && end > start) {
+                return trimmed.substring(start + 1, end);
+            }
+            break;
+        }
+        return null;
+    }
+
+    private static class ColumnInfo {
+        final String name;
+        final int order;
+        ColumnInfo(String name, int order) {
+            this.name = name;
+            this.order = order;
+        }
+    }
+
+}
diff --git a/cwms-data-api/src/test/java/cwms/cda/api/TimeseriesControllerTestIT.java b/cwms-data-api/src/test/java/cwms/cda/api/TimeseriesControllerTestIT.java
index a4bf226dc0..f2de16bcb1 100644
--- a/cwms-data-api/src/test/java/cwms/cda/api/TimeseriesControllerTestIT.java
+++ b/cwms-data-api/src/test/java/cwms/cda/api/TimeseriesControllerTestIT.java
@@ -15,7 +15,9 @@
 import static cwms.cda.api.Controllers.TRIM;
 import static cwms.cda.api.Controllers.UNIT;
 import static cwms.cda.api.Controllers.VERSION_DATE;
+import static cwms.cda.api.TimeSeriesController.DEFAULT_PAGE_SIZE;
 import static cwms.cda.data.dao.JooqDao.getDslContext;
+import static cwms.cda.formatters.DateFormatParameter.CUSTOM;
 import static cwms.cda.helpers.DatabaseHelpers.LATEST_SCHEMA;
 import static helpers.FloatCloseTo.floatCloseTo;
 import static io.restassured.RestAssured.given;
@@ -26,6 +28,8 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -62,7 +66,6 @@
 import javax.servlet.http.HttpServletResponse;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
-import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
@@ -1822,10 +1825,662 @@ void test_wrong_units() throws Exception {
             .log().ifValidationFails(LogDetail.ALL,true)
         .assertThat()
             .statusCode(is(HttpServletResponse.SC_BAD_REQUEST))
-            .body("details.message", containsString("Cannot convert from unit C to unit m"))
+            .body("details.message", containsString("m is not a valid unit for time series Buckhorn.Temp-Water.Inst.1Day.0.cda-test"))
         ;
     }
 
+    @Test
+    void test_csv_default_columns_with_metadata() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/1day_offset.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+        String firstPoint = "2023-02-02T06:00:00-05:00";
+        // insert
+        given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        // retrieve CSV v2 with metadata comments explicitly requested
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+            .queryParam(Controllers.INCLUDE_METADATA_AS_CSV_COMMENTS, true)
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        assertTrue(normalized.startsWith("# "), "Expected metadata comments");
+        assertTrue(normalized.contains("date-time,value (F)\n"),
+                "Expected header 'date-time,value (F)' but was: " + normalized.split("\n")[0]);
+    }
+
+
+    @Test
+    void test_csv_optional_columns_no_metadata() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/1day_offset.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+        String firstPoint = "2023-02-02T06:00:00-05:00";
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+            .queryParam("include-optional-csv-columns", true)
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        assertTrue(normalized.contains("date-time,value (F),data-entry-date,quality-code"),
+                "Expected header with optional columns but was: " + normalized.split("\n")[0]);
+    }
+
+    @Test
+    void test_csv_optional_columns_with_metadata() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/1day_offset.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+        String firstPoint = "2023-02-02T06:00:00-05:00";
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+            .queryParam("include-metadata-as-comments", true)
+            .queryParam("include-optional-csv-columns", true)
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        assertTrue(normalized.startsWith("# time-series-id: "), "Expected metadata comments");
+        assertTrue(normalized.contains("date-time,value (F),data-entry-date,quality-code"),
+                "Expected header with optional columns");
+    }
+
+    @Test
+    void test_csv_streaming_no_missing_data() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/1day_offset_version_date_max.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        // inserting the time series
+        given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .redirects().follow(true)
+            .redirects().max(3)
+            .post("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+        .assertThat()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String secondVersionDate = "1604786000000";
+        tsData = tsData.replace("1594786000000", secondVersionDate).replace("35,", "47.5,");
+        // inserting the second time series
+        given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .redirects().follow(true)
+            .redirects().max(3)
+            .post("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+        .assertThat()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        // get it back
+        String firstPoint = "2023-02-02T06:00:00-05:00"; //aka 2023-02-02T11:00:00.000Z or
+        String versionDate = "2020-07-15T04:06:40Z";
+
+        // 2. Client requests CSV with metadata and optional columns
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(OFFICE, officeId)
+            .queryParam(UNIT, "F")
+            .queryParam(NAME, ts.get(NAME).asText())
+            .queryParam(BEGIN, firstPoint)
+            .queryParam(END, firstPoint)
+            .queryParam(VERSION_DATE, versionDate)
+            .queryParam(Controllers.INCLUDE_METADATA_AS_CSV_COMMENTS, "true")
+            .queryParam(Controllers.INCLUDE_OPTIONAL_CSV_COLUMNS, "true")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        String[] lines = normalized.split("\n");
+
+        // 3. Verify Metadata
+        // Expected metadata fields (from CsvV1 or similar)
+        String[] expectedMetadata = {
+            "time-series-id", "office-id", "version-date",
+        };
+
+        for (String key : expectedMetadata) {
+            boolean found = false;
+            for (String line : lines) {
+                if (line.startsWith("# " + key + ":")) {
+                    found = true;
+                    String value = line.substring(line.indexOf(":") + 1).trim();
+                    assertFalse(value.isEmpty(), "Metadata field " + key + " should not be empty");
+                    break;
+                }
+            }
+            assertTrue(found, "Metadata field " + key + " was not found in response");
+        }
+
+        // 4. Verify Header and Data Row
+        int headerIndex = -1;
+        for (int i = 0; i < lines.length; i++) {
+            if (!lines[i].startsWith("#") && !lines[i].trim().isEmpty()) {
+                headerIndex = i;
+                break;
+            }
+        }
+        assertTrue(headerIndex != -1, "CSV header not found");
+        assertEquals("date-time,value (F),data-entry-date,quality-code", lines[headerIndex]);
+
+        String firstDataRow = lines[headerIndex + 1];
+        String[] columns = firstDataRow.split(",");
+        assertEquals(4, columns.length, "Expected 4 columns in data row: " + firstDataRow);
+
+        for (int i = 0; i < columns.length; i++) {
+            assertFalse(columns[i].trim().isEmpty(), "Column " + i + " in first data row should not be empty. Row: " + firstDataRow);
+        }
+    }
+
+    @Test
+    void test_csv_streaming_with_optionals() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/large_timeseries.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        // 1. Store the large dataset
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization", user.toHeaderValue())
+            .queryParam(OFFICE, officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        // 2. Client requests CSV with optional columns and receives a single streamed response.
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.PAGE_SIZE, 10)
+            .queryParam(Controllers.BEGIN, "2023-01-11T12:00:00Z")
+            .queryParam(Controllers.END, "2023-01-13T13:00:00Z")
+            .queryParam(Controllers.INCLUDE_OPTIONAL_CSV_COLUMNS, "true")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        // 3. Verify total data integrity
+        assertNotNull(body);
+        String[] lines = body.replace("\r", "").split("\n");
+        // 50 points + 1 header = 51 lines
+        assertEquals(51, lines.length, "Expected 51 lines (1 header + 50 points)");
+        assertEquals("date-time,value (cfs),data-entry-date,quality-code", lines[0]);
+
+        // Verify first and last data points
+        // Data entry date is non-null for stored points, and quality code should be present.
+        // Format: 2023-01-11T12:00:00Z,10.0,2023-01-11T12:00:00Z,0
+        // (Assuming data entry date is approximately now, but for these tests it might be same as date-time if not specified)
+        assertTrue(lines[1].startsWith("2023-01-11T12:00:00Z,10.0"), "Row 1 mismatch: " + lines[1]);
+        assertTrue(lines[1].endsWith(",0"), "Row 1 quality code mismatch: " + lines[1]);
+        assertTrue(lines[50].startsWith("2023-01-13T13:00:00Z,500.0"), "Row 50 mismatch: " + lines[50]);
+        assertTrue(lines[50].endsWith(",0"), "Row 50 quality code mismatch: " + lines[50]);
+    }
+
+    @Test
+    void test_csv_streaming() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/large_timeseries.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        // 1. Store the large dataset
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization", user.toHeaderValue())
+            .queryParam(OFFICE, officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        // 2. Client requests CSV and receives a single streamed response.
+        // The small page-size (10) forces the server to perform 5 internal fetch cycles
+        // for the 50 points in large_timeseries.json.
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.PAGE_SIZE, 10)
+            .queryParam(Controllers.BEGIN, "2023-01-11T12:00:00Z")
+            .queryParam(Controllers.END, "2023-01-13T13:00:00Z")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .header("X-Stream-Batch-Size", "10")
+            .extract().asString();
+
+        // 3. Verify total data integrity
+        assertNotNull(body);
+        String[] lines = body.replace("\r", "").split("\n");
+        // 50 points + 1 header = 51 lines
+        assertEquals(51, lines.length, "Expected 51 lines (1 header + 50 points)");
+        assertEquals("date-time,value (cfs)", lines[0]);
+
+        // Verify first and last data points
+        assertTrue(lines[1].startsWith("2023-01-11T12:00:00Z,10.0"));
+        assertTrue(lines[50].startsWith("2023-01-13T13:00:00Z,500.0"));
+
+        // 4. Client requests CSV and receives a single streamed response.
+        // The small page-size (6) forces the server to perform 9 internal fetch cycles
+        // for the 50 points in large_timeseries.json.
+        body = given()
+                .log().ifValidationFails(LogDetail.ALL, true)
+                .accept(Formats.CSV)
+                .queryParam(Controllers.OFFICE, officeId)
+                .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+                .queryParam(Controllers.PAGE_SIZE, 6)
+                .queryParam(Controllers.BEGIN, "2023-01-11T12:00:00Z")
+                .queryParam(Controllers.END, "2023-01-13T13:00:00Z")
+            .when()
+                .get("/timeseries/")
+            .then()
+                .log().ifValidationFails(LogDetail.ALL, true)
+                .statusCode(is(HttpServletResponse.SC_OK))
+                .header("X-Stream-Batch-Size", "6")
+                .extract().asString();
+
+        // 5. Verify total data integrity
+        assertNotNull(body);
+        lines = body.replace("\r", "").split("\n");
+        // 50 points + 1 header = 51 lines
+        assertEquals(51, lines.length, "Expected 51 lines (1 header + 50 points)");
+        assertEquals("date-time,value (cfs)", lines[0]);
+
+        // Verify first and last data points
+        assertTrue(lines[1].startsWith("2023-01-11T12:00:00Z,10.0"));
+        assertTrue(lines[50].startsWith("2023-01-13T13:00:00Z,500.0"));
+
+        // 6. Client requests CSV and receives a single streamed response.
+        // The large page-size (60) forces the server to perform 1 internal fetch cycles
+        // for the 50 points in large_timeseries.json.
+        body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.PAGE_SIZE, 60)
+            .queryParam(Controllers.BEGIN, "2023-01-11T12:00:00Z")
+            .queryParam(Controllers.END, "2023-01-13T13:00:00Z")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .header("X-Stream-Batch-Size", "60")
+            .extract().asString();
+
+        // 7. Verify total data integrity
+        assertNotNull(body);
+        lines = body.replace("\r", "").split("\n");
+        // 50 points + 1 header = 51 lines
+        assertEquals(51, lines.length, "Expected 51 lines (1 header + 50 points)");
+        assertEquals("date-time,value (cfs)", lines[0]);
+
+        // Verify first and last data points
+        assertTrue(lines[1].startsWith("2023-01-11T12:00:00Z,10.0"));
+        assertTrue(lines[50].startsWith("2023-01-13T13:00:00Z,500.0"));
+    }
+
+    @Test
+    void test_query_param_format_overrides_header() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/large_timeseries.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        // 1. Store the large dataset
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization", user.toHeaderValue())
+            .queryParam(OFFICE, officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        //this should use csv not json because the query param overrides the header
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.JSON)
+            .queryParam(Controllers.FORMAT, Formats.CSV_LEGACY)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, "2023-01-11T12:00:00Z")
+            .queryParam(Controllers.END, "2023-01-13T13:00:00Z")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .header("X-Stream-Batch-Size", String.valueOf(DEFAULT_PAGE_SIZE))
+            .extract().asString();
+
+        // 3. Verify total data integrity
+        assertNotNull(body);
+        String[] lines = body.replace("\r", "").split("\n");
+        // 50 points + 1 header = 51 lines
+        assertEquals(51, lines.length, "Expected 51 lines (1 header + 50 points)");
+        assertEquals("date-time,value (cfs)", lines[0]);
+
+        // Verify first and last data points
+        assertTrue(lines[1].startsWith("2023-01-11T12:00:00Z,10.0"));
+        assertTrue(lines[50].startsWith("2023-01-13T13:00:00Z,500.0"));
+    }
+
+    @Test
+    void test_csv_default_columns_no_metadata() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream(
+                "/cwms/cda/api/lrl/1day_offset.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+        String firstPoint = "2023-02-02T06:00:00-05:00";
+
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization",user.toHeaderValue())
+            .queryParam(OFFICE,officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL,true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        assertFalse(normalized.startsWith("# "), "Expected no metadata comments");
+        assertTrue(normalized.startsWith("date-time,value (F)\n"),
+                "Expected default columns header but was: " + normalized.split("\n")[0]);
+    }
+
+    @Test
+    void test_csv_date_format_query_param() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream("/cwms/cda/api/lrl/1hour.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization", user.toHeaderValue())
+            .queryParam(OFFICE, officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String firstPoint = "2023-01-11T12:00:00Z";
+
+        // Test with date-format param in query string
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+            .queryParam(Controllers.DATE_FORMAT, "epoch-millis")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        // epoch-millis for 2023-01-11T12:00:00Z is 1673438400000
+        assertTrue(normalized.contains("1673438400000"), "Expected epoch-millis date format via query param but was: " + normalized);
+    }
+
+    @Test
+    void test_csv_date_format_custom_pattern() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        InputStream resource = this.getClass().getResourceAsStream("/cwms/cda/api/lrl/1hour.json");
+        assertNotNull(resource);
+        String tsData = IOUtils.toString(resource, StandardCharsets.UTF_8);
+
+        JsonNode ts = mapper.readTree(tsData);
+        String location = ts.get(NAME).asText().split("\\.")[0];
+        String officeId = ts.get("office-id").asText();
+
+        createLocation(location, true, officeId);
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+
+        given()
+            .accept(Formats.JSONV2)
+            .contentType(Formats.JSONV2)
+            .body(tsData)
+            .header("Authorization", user.toHeaderValue())
+            .queryParam(OFFICE, officeId)
+        .when()
+            .post("/timeseries/")
+        .then()
+            .statusCode(is(HttpServletResponse.SC_OK));
+
+        String firstPoint = "2023-01-11T12:00:00Z";
+
+        // Test with custom date-format-pattern
+        String body = given()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .accept(Formats.CSV)
+            .queryParam(Controllers.OFFICE, officeId)
+            .queryParam(Controllers.NAME, ts.get(Controllers.NAME).asText())
+            .queryParam(Controllers.BEGIN, firstPoint)
+            .queryParam(Controllers.END, firstPoint)
+            .queryParam(Controllers.DATE_FORMAT, CUSTOM)
+            .queryParam(Controllers.DATE_FORMAT_PATTERN, "yyyyMMddHHmm")
+        .when()
+            .get("/timeseries/")
+        .then()
+            .log().ifValidationFails(LogDetail.ALL, true)
+            .statusCode(is(HttpServletResponse.SC_OK))
+            .extract().asString();
+
+        assertNotNull(body);
+        String normalized = body.replace("\r", "");
+        // 2023-01-11T12:00:00Z -> 202301111200 with pattern "yyyyMMddHHmm"
+        assertTrue(normalized.contains("202301111200"), "Expected custom date format pattern but was: " + normalized);
+    }
+
     enum GetAllTest
     {
         DEFAULT(Formats.DEFAULT, Formats.JSONV2),
diff --git a/cwms-data-api/src/test/java/cwms/cda/data/dto/csv/TestTimeSeriesCsvRow.java b/cwms-data-api/src/test/java/cwms/cda/data/dto/csv/TestTimeSeriesCsvRow.java
new file mode 100644
index 0000000000..44f08d63ba
--- /dev/null
+++ b/cwms-data-api/src/test/java/cwms/cda/data/dto/csv/TestTimeSeriesCsvRow.java
@@ -0,0 +1,259 @@
+package cwms.cda.data.dto.csv;
+
+import cwms.cda.formatters.DateFormatResolver;
+import cwms.cda.formatters.DateFormat;
+import cwms.cda.formatters.csv.CsvConfiguration;
+import cwms.cda.formatters.csv.CsvV1;
+import org.apache.commons.io.IOUtils;
+import org.junit.jupiter.api.Test;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+
+import static cwms.cda.helpers.DTOMatch.assertMatch;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestTimeSeriesCsvRow {
+
+    @Test
+    void testSingleRow_Default_Columns_NoMetadata() throws Exception {
+        TimeSeriesCsvRow row = buildRow(Instant.parse("2021-06-21T21:06:00Z"), 1.0, "ft");
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withRows(Collections.singletonList(row))
+                .build();
+
+        CsvV1 csv = new CsvV1();
+        String actual = csv.format(container);
+        assertNotNull(actual);
+        String normalized = normalize(actual);
+        assertTrue(normalized.contains("date-time,value (ft)"), "Header mismatch");
+        assertTrue(normalized.contains("2021-06-21T21:06:00Z,1.0"), "Row mismatch");
+    }
+
+    @Test
+    void testSingleRow_Default_Columns_WithMetadataComments() throws Exception {
+        TimeSeriesCsvRow row = buildRow(Instant.parse("2021-06-21T21:06:00Z"), 1.0, "ft");
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withTimeSeriesId("RYAN3.Stage.Inst.5Minutes.0.ZSTORE_TS_TEST")
+                .withOfficeId("SPK")
+                .withVersionDate("2025-07-22T14:00:00Z")
+                .withRows(Collections.singletonList(row))
+                .build();
+
+        CsvV1 csv = new CsvV1();
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withMetadataIncluded(true)
+                .withOptionalColumnsIncluded(false)
+                .withDateFormat(DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN))
+                .build();
+        String actual = csv.format(container, config);
+        assertNotNull(actual);
+        String normalized = normalize(actual);
+        assertTrue(normalized.contains("# time-series-id: RYAN3.Stage.Inst.5Minutes.0.ZSTORE_TS_TEST"), "Metadata mismatch");
+        assertTrue(normalized.contains("# office-id: SPK"), "Metadata mismatch");
+        assertTrue(normalized.contains("date-time,value (ft)"), "Header mismatch");
+        assertTrue(normalized.contains("2021-06-21T21:06:00Z,1.0"), "Row mismatch");
+    }
+
+    @Test
+    void testSingleRow_WithOptionalColumns() {
+        TimeSeriesCsvRow row = new TimeSeriesCsvRow.Builder()
+                .withDateTime(Instant.parse("2021-06-21T21:06:00Z"))
+                .withValue(1.0)
+                .withQualityCode(0)
+                .withDataEntryDate(Instant.parse("2021-06-21T21:00:00Z"))
+                .withUnits("ft")
+                .build();
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withRows(Collections.singletonList(row))
+                .build();
+
+        CsvV1 csv = new CsvV1();
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withOptionalColumnsIncluded(true)
+                .build();
+        String actual = csv.format(container, config);
+        assertNotNull(actual);
+        String normalized = normalize(actual);
+        assertTrue(normalized.contains("date-time,value (ft),data-entry-date,quality-code"), "Header mismatch: " + normalized);
+        assertTrue(normalized.contains("2021-06-21T21:06:00Z,1.0,2021-06-21T21:00:00Z,0"), "Row mismatch: " + normalized);
+    }
+
+    @Test
+    void testDefaultSerialization() throws Exception {
+        String csv = readResource("cwms/cda/data/dto/time-series-default.csv");
+        CsvV1 formatter = new CsvV1();
+        TimeSeriesCsv container = formatter.parseContent(csv, TimeSeriesCsv.class);
+        assertNotNull(container);
+        List<TimeSeriesCsvRow> rows = container.getRows();
+        assertEquals(4, rows.size());
+        assertEquals("cfs", rows.get(0).getUnits());
+        assertEquals(0.0, rows.get(0).getValue());
+        assertEquals(Instant.parse("2021-06-21T00:00:00Z"), rows.get(0).getDateTime());
+
+        //serialize back
+        String serialized = formatter.format(container);
+        assertNotNull(serialized);
+
+        //parse back
+        TimeSeriesCsv parsedContainer = formatter.parseContent(serialized, TimeSeriesCsv.class);
+        assertNotNull(parsedContainer);
+        assertEquals(rows.get(0).getUnits(), parsedContainer.getRows().get(0).getUnits());
+        assertMatch(rows, parsedContainer.getRows());
+    }
+
+    @Test
+    void testMetadataAsCommentsSerialization() throws Exception {
+        String csv = readResource("cwms/cda/data/dto/time-series-metadata-comments.csv");
+        CsvV1 formatter = new CsvV1();
+        TimeSeriesCsv container = formatter.parseContent(csv, TimeSeriesCsv.class);
+        assertNotNull(container);
+        List<TimeSeriesCsvRow> rows = container.getRows();
+        assertEquals(4, rows.size());
+        assertEquals("cfs", rows.get(0).getUnits());
+        assertEquals(0.0, rows.get(0).getValue());
+        assertEquals("ALAT2.Flow-Out.Inst.1Hour.0.Rev-SWF-REGI", container.getTimeSeriesId());
+        assertEquals("SWT", container.getOfficeId());
+        assertEquals("aggregate", container.getVersionDate());
+
+        //serialize back
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withMetadataIncluded(true)
+                .withOptionalColumnsIncluded(false)
+                .withDateFormat(DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN))
+                .build();
+        String serialized = formatter.format(container, config);
+        assertNotNull(serialized);
+
+        //parse back
+        TimeSeriesCsv parsedContainer = formatter.parseContent(serialized, TimeSeriesCsv.class);
+        assertNotNull(parsedContainer);
+        assertEquals(container.getTimeSeriesId(), parsedContainer.getTimeSeriesId());
+        assertEquals(container.getOfficeId(), parsedContainer.getOfficeId());
+        assertEquals(container.getVersionDate(), parsedContainer.getVersionDate());
+        assertMatch(rows, parsedContainer.getRows());
+    }
+
+    @Test
+    void testOptionalsNoMetadataSerialization() throws Exception {
+        String csv = readResource("cwms/cda/data/dto/time-series-optionals-no-metadata-comments.csv");
+        CsvV1 formatter = new CsvV1();
+        TimeSeriesCsv container = formatter.parseContent(csv, TimeSeriesCsv.class);
+        assertNotNull(container);
+        List<TimeSeriesCsvRow> rows = container.getRows();
+        assertEquals(4, rows.size());
+        assertEquals("cfs", rows.get(0).getUnits());
+        assertEquals(0.0, rows.get(0).getValue());
+        assertNotNull(rows.get(0).getDataEntryDate());
+        assertEquals(5, rows.get(0).getQualityCode());
+
+        //serialize back
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withOptionalColumnsIncluded(true)
+                .build();
+        String serialized = formatter.format(container, config);
+        assertNotNull(serialized);
+
+        //parse back
+        TimeSeriesCsv parsedContainer = formatter.parseContent(serialized, TimeSeriesCsv.class);
+        assertNotNull(parsedContainer);
+        assertMatch(rows, parsedContainer.getRows());
+    }
+
+    @Test
+    void testOptionalsWithMetadataCommentsSerialization() throws Exception {
+        String csv = readResource("cwms/cda/data/dto/time-series-optionals-with-metadata-comments.csv");
+        CsvV1 formatter = new CsvV1();
+        TimeSeriesCsv container = formatter.parseContent(csv, TimeSeriesCsv.class);
+        assertNotNull(container);
+        List<TimeSeriesCsvRow> rows = container.getRows();
+        assertEquals(4, rows.size());
+        assertEquals("cfs", rows.get(0).getUnits());
+        assertEquals(0.0, rows.get(0).getValue());
+        assertEquals("ALAT2.Flow-Out.Inst.1Hour.0.Rev-SWF-REGI", container.getTimeSeriesId());
+        assertEquals("SWT", container.getOfficeId());
+        assertEquals("aggregate", container.getVersionDate());
+        assertNotNull(rows.get(0).getDataEntryDate());
+        assertEquals(5, rows.get(0).getQualityCode());
+
+        //serialize back
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withMetadataIncluded(true)
+                .withOptionalColumnsIncluded(true)
+                .withDateFormat(DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN))
+                .build();
+        String serialized = formatter.format(container, config);
+        assertNotNull(serialized);
+
+        //parse back
+        TimeSeriesCsv parsedContainer = formatter.parseContent(serialized, TimeSeriesCsv.class);
+        assertNotNull(parsedContainer);
+        assertEquals(container.getTimeSeriesId(), parsedContainer.getTimeSeriesId());
+        assertEquals(container.getOfficeId(), parsedContainer.getOfficeId());
+        assertEquals(container.getVersionDate(), parsedContainer.getVersionDate());
+        assertMatch(rows, parsedContainer.getRows());
+    }
+
+    @Test
+    void testSingleRow_EpochMillis() throws Exception {
+        TimeSeriesCsvRow row = buildRow(Instant.parse("2021-06-21T21:06:00Z"), 1.0, "ft");
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withRows(Collections.singletonList(row))
+                .build();
+
+        CsvV1 csv = new CsvV1();
+        cwms.cda.formatters.csv.CsvConfiguration config = new cwms.cda.formatters.csv.CsvConfiguration.Builder()
+                .withDateFormat(DateFormat.epochMillis())
+                .build();
+        String actual = csv.format(container, config);
+        assertNotNull(actual);
+        String normalized = normalize(actual);
+        assertTrue(normalized.contains("date-time,value (ft)"), "Header mismatch");
+        // 2021-06-21T21:06:00Z is 1624309560000
+        assertTrue(normalized.contains("1624309560000,1.0"), "Row mismatch: " + normalized);
+    }
+
+    @Test
+    void testSingleRow_CustomPattern() throws Exception {
+        TimeSeriesCsvRow row = buildRow(Instant.parse("2021-06-21T21:06:00Z"), 1.0, "ft");
+        TimeSeriesCsv container = new TimeSeriesCsv.Builder()
+                .withRows(Collections.singletonList(row))
+                .build();
+
+        CsvV1 csv = new CsvV1();
+        CsvConfiguration config = new CsvConfiguration.Builder()
+                .withDateFormat(DateFormat.pattern("yyyyMMddHHmm"))
+                .build();
+        String actual = csv.format(container, config);
+        assertNotNull(actual);
+        String normalized = normalize(actual);
+        assertTrue(normalized.contains("date-time,value (ft)"), "Header mismatch");
+        // 2021-06-21T21:06:00Z should be 202106212106
+        assertTrue(normalized.contains("202106212106,1.0"), "Row mismatch: " + normalized);
+    }
+
+    private static String readResource(String path) throws Exception {
+        InputStream stream = TestTimeSeriesCsvRow.class.getClassLoader().getResourceAsStream(path);
+        assertNotNull(stream, "Missing test resource: " + path);
+        String expected = IOUtils.toString(stream, StandardCharsets.UTF_8);
+        return normalize(expected);
+    }
+
+    private static String normalize(String s) {
+        return s.replaceAll("\r", "");
+    }
+
+    private static TimeSeriesCsvRow buildRow(Instant dateTime, Double value, String units) {
+        return new TimeSeriesCsvRow.Builder()
+                .withDateTime(dateTime)
+                .withValue(value)
+                .withQualityCode(0)
+                .withUnits(units)
+                .build();
+    }
+}
diff --git a/cwms-data-api/src/test/java/cwms/cda/formatters/DateFormatResolverTest.java b/cwms-data-api/src/test/java/cwms/cda/formatters/DateFormatResolverTest.java
new file mode 100644
index 0000000000..72956e713f
--- /dev/null
+++ b/cwms-data-api/src/test/java/cwms/cda/formatters/DateFormatResolverTest.java
@@ -0,0 +1,49 @@
+package cwms.cda.formatters;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import cwms.cda.api.Controllers;
+import io.javalin.http.Context;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class DateFormatResolverTest {
+
+    @Test
+    void testResolve_Default() {
+        assertEquals(DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN), DateFormatResolver.resolve(null, null));
+    }
+
+    @Test
+    void testResolve_Enum() {
+        assertEquals(DateFormat.epochMillis(), DateFormatResolver.resolve(DateFormatParameter.EPOCH_MILLIS.getValue(), null));
+
+        assertEquals(DateFormat.pattern(DateFormatResolver.ISO_INSTANT_PATTERN), DateFormatResolver.resolve(DateFormatParameter.ISO_INSTANT.getValue(), null));
+
+        assertEquals(DateFormat.pattern(DateFormatResolver.ISO_OFFSET_PATTERN), DateFormatResolver.resolve(DateFormatParameter.ISO_OFFSET.getValue(), null));
+        
+        assertEquals(DateFormat.pattern(DateFormatResolver.ISO_LOCAL_PATTERN), DateFormatResolver.resolve(DateFormatParameter.ISO_LOCAL.getValue(), null));
+
+        assertEquals(DateFormat.pattern(DateFormatResolver.DATE_ONLY_PATTERN), DateFormatResolver.resolve(DateFormatParameter.DATE_ONLY.getValue(), null));
+    }
+
+    @Test
+    void testResolve_CustomWithPattern() {
+        assertEquals(DateFormat.pattern("yyyyMMdd"), DateFormatResolver.resolve("custom", "yyyyMMdd"));
+    }
+
+    @Test
+    void testResolve_CustomMissingPattern() {
+        assertThrows(IllegalArgumentException.class, () -> DateFormatResolver.resolve("custom", null));
+    }
+
+    @Test
+    void testResolve_Unsupported() {
+        assertThrows(UnsupportedOperationException.class, () -> DateFormatResolver.resolve("unsupported-format", null));
+    }
+
+}
diff --git a/cwms-data-api/src/test/java/cwms/cda/formatters/FormatsTest.java b/cwms-data-api/src/test/java/cwms/cda/formatters/FormatsTest.java
index a176d2ce00..266e00b69b 100644
--- a/cwms-data-api/src/test/java/cwms/cda/formatters/FormatsTest.java
+++ b/cwms-data-api/src/test/java/cwms/cda/formatters/FormatsTest.java
@@ -106,21 +106,6 @@ void testParseHeaderAndQueryParmXml() {
 
     }
 
-    @Test
-    void testParseBoth() {
-        assertThrows(FormattingException.class, () -> {
-            Formats.parseHeaderAndQueryParm("application/json", "json", LocationLevels.class);
-        });
-    }
-
-    @Test
-    void testParseBothv2() {
-        assertThrows(FormattingException.class, () -> {
-            Formats.parseHeaderAndQueryParm("application/json;version=2", "json", LocationLevels.class);
-        });
-
-    }
-
     @Test
     void testParseHeader() {
         ContentType contentType;
diff --git a/cwms-data-api/src/test/java/cwms/cda/helpers/DTOMatch.java b/cwms-data-api/src/test/java/cwms/cda/helpers/DTOMatch.java
index 44ca67f119..11ed2329f6 100644
--- a/cwms-data-api/src/test/java/cwms/cda/helpers/DTOMatch.java
+++ b/cwms-data-api/src/test/java/cwms/cda/helpers/DTOMatch.java
@@ -34,6 +34,8 @@
 import cwms.cda.data.dto.LocationToPublishedData;
 import cwms.cda.data.dto.LocationToPublishedDataList;
 import cwms.cda.data.dto.PublishedTimeSeriesData;
+import cwms.cda.data.dto.csv.TimeSeriesCsv;
+import cwms.cda.data.dto.csv.TimeSeriesCsvRow;
 import cwms.cda.data.dto.location.kind.Lock;
 import cwms.cda.data.dto.CwmsDTOBase;
 import cwms.cda.data.dto.location.kind.GateChange;
@@ -407,6 +409,29 @@ public static void assertMatch(WaterSupplyAccounting first, WaterSupplyAccountin
         );
     }
 
+    public static void assertMatch(TimeSeriesCsv first, TimeSeriesCsv second) {
+        assertAll(
+            () -> assertEquals(first.getTimeSeriesId(), second.getTimeSeriesId(), "Time series IDs do not match"),
+            () -> assertEquals(first.getOfficeId(), second.getOfficeId(), "Office IDs do not match"),
+            () -> assertEquals(first.getVersionDate(), second.getVersionDate(), "Version dates do not match"),
+            () -> assertMatch(first.getRows(), second.getRows())
+        );
+    }
+
+    public static void assertMatch(List<TimeSeriesCsvRow> first, List<TimeSeriesCsvRow> second) {
+        assertMatch(first, second, DTOMatch::assertMatch);
+    }
+
+    public static void assertMatch(TimeSeriesCsvRow first, TimeSeriesCsvRow second) {
+        assertAll(
+            () -> assertEquals(first.getDateTime(), second.getDateTime(), "Date times do not match"),
+            () -> assertEquals(first.getDataEntryDate(), second.getDataEntryDate(), "Data entry dates do not match"),
+            () -> assertEquals(first.getUnits(), second.getUnits(), "Units do not match"),
+            () -> assertEquals(first.getValue(), second.getValue(), "Values do not match"),
+            () -> assertEquals(first.getQualityCode(), second.getQualityCode(), "Quality codes do not match")
+        );
+    }
+
     public static void assertMatch(Map<Instant, List<PumpTransfer>> first, Map<Instant, List<PumpTransfer>> second) {
         assertAll(
             () -> assertEquals(first.size(), second.size(), "Pump accounting sizes do not match"),
diff --git a/cwms-data-api/src/test/resources/cwms/cda/api/lrl/large_timeseries.json b/cwms-data-api/src/test/resources/cwms/cda/api/lrl/large_timeseries.json
new file mode 100644
index 0000000000..cc3810420b
--- /dev/null
+++ b/cwms-data-api/src/test/resources/cwms/cda/api/lrl/large_timeseries.json
@@ -0,0 +1,57 @@
+{
+    "name": "Calhoun.Flow.Inst.1Hour.0.cda-test",
+    "office-id": "SPK",
+    "units": "cfs",
+    "values": [
+        [1673438400000, 10, 0],
+        [1673442000000, 20, 0],
+        [1673445600000, 30, 0],
+        [1673449200000, 40, 0],
+        [1673452800000, 50, 0],
+        [1673456400000, 60, 0],
+        [1673460000000, 70, 0],
+        [1673463600000, 80, 0],
+        [1673467200000, 90, 0],
+        [1673470800000, 100, 0],
+        [1673474400000, 110, 0],
+        [1673478000000, 120, 0],
+        [1673481600000, 130, 0],
+        [1673485200000, 140, 0],
+        [1673488800000, 150, 0],
+        [1673492400000, 160, 0],
+        [1673496000000, 170, 0],
+        [1673499600000, 180, 0],
+        [1673503200000, 190, 0],
+        [1673506800000, 200, 0],
+        [1673510400000, 210, 0],
+        [1673514000000, 220, 0],
+        [1673517600000, 230, 0],
+        [1673521200000, 240, 0],
+        [1673524800000, 250, 0],
+        [1673528400000, 260, 0],
+        [1673532000000, 270, 0],
+        [1673535600000, 280, 0],
+        [1673539200000, 290, 0],
+        [1673542800000, 300, 0],
+        [1673546400000, 310, 0],
+        [1673550000000, 320, 0],
+        [1673553600000, 330, 0],
+        [1673557200000, 340, 0],
+        [1673560800000, 350, 0],
+        [1673564400000, 360, 0],
+        [1673568000000, 370, 0],
+        [1673571600000, 380, 0],
+        [1673575200000, 390, 0],
+        [1673578800000, 400, 0],
+        [1673582400000, 410, 0],
+        [1673586000000, 420, 0],
+        [1673589600000, 430, 0],
+        [1673593200000, 440, 0],
+        [1673596800000, 450, 0],
+        [1673600400000, 460, 0],
+        [1673604000000, 470, 0],
+        [1673607600000, 480, 0],
+        [1673611200000, 490, 0],
+        [1673614800000, 500, 0]
+    ]
+}
diff --git a/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-default.csv b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-default.csv
new file mode 100644
index 0000000000..75bdb32b51
--- /dev/null
+++ b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-default.csv
@@ -0,0 +1,5 @@
+date-time, value (cfs)
+2021-06-21T00:00:00Z, 0.0
+2021-06-22T00:00:00Z, 1.0
+2021-06-23T00:00:00Z, 2.0
+2021-06-24T00:00:00Z, 3.0
diff --git a/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-metadata-comments.csv b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-metadata-comments.csv
new file mode 100644
index 0000000000..50e2a69716
--- /dev/null
+++ b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-metadata-comments.csv
@@ -0,0 +1,8 @@
+# time-series-id: ALAT2.Flow-Out.Inst.1Hour.0.Rev-SWF-REGI
+# office-id: SWT
+# version-date: aggregate
+date-time, value (cfs)
+2021-06-21T00:00:00Z, 0.0
+2021-06-22T00:00:00Z, 1.0
+2021-06-23T00:00:00Z, 2.0
+2021-06-24T00:00:00Z, 3.0
diff --git a/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-no-metadata-comments.csv b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-no-metadata-comments.csv
new file mode 100644
index 0000000000..c509376e00
--- /dev/null
+++ b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-no-metadata-comments.csv
@@ -0,0 +1,5 @@
+date-time, value (cfs), data-entry-date, quality-code
+2021-06-21T00:00:00Z, 0.0, 2021-06-21T00:05:00Z, 5
+2021-06-22T00:00:00Z, 1.0, 2021-06-22T00:05:00Z, 5
+2021-06-23T00:00:00Z, 2.0, 2021-06-23T00:05:00Z, 5
+2021-06-24T00:00:00Z, 3.0, 2021-06-24T00:05:00Z, 5
diff --git a/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-with-metadata-comments.csv b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-with-metadata-comments.csv
new file mode 100644
index 0000000000..13ec27f224
--- /dev/null
+++ b/cwms-data-api/src/test/resources/cwms/cda/data/dto/time-series-optionals-with-metadata-comments.csv
@@ -0,0 +1,9 @@
+# time-series-id: ALAT2.Flow-Out.Inst.1Hour.0.Rev-SWF-REGI
+# office-id: SWT
+# units: cfs
+# version-date: aggregate
+date-time,value (cfs),data-entry-date,quality-code
+2021-06-21T00:00:00Z,0.0,2021-06-21T00:05:00Z,5
+2021-06-21T01:00:00Z,10.0,2021-06-21T01:05:00Z,5
+2021-06-21T02:00:00Z,20.0,2021-06-21T02:05:00Z,5
+2021-06-21T03:00:00Z,30.0,2021-06-21T03:05:00Z,5
diff --git a/docs/source/decisions/0008-timeseries-csv-format.rst b/docs/source/decisions/0008-timeseries-csv-format.rst
index f021e461db..53ad4ef594 100644
--- a/docs/source/decisions/0008-timeseries-csv-format.rst
+++ b/docs/source/decisions/0008-timeseries-csv-format.rst
@@ -38,7 +38,7 @@ Key points
      - Optional (off by default): ``quality-code``, ``data-entry-date``
      - Because headers are always included, optional columns can be toggled without breaking parsing. Clients should rely on column names, not indices. Given units are in the `value` header, clients will need to handle this appropriately to determine the correct column index.
    * - Metadata fields
-     - Emitted as top-of-payload comments (``metadata-format=comments``)
+     - Emitted as top-of-payload comments if query parameter is set to include (``include-metadata-as-comments=true`` (default = false)), otherwise omitted.
      - The following fields can be treated as metadata comments at top-of-payload: ``time-series-id``, ``office-id``, ``version-date``. These are optional (off by default). It is assumed that the only comments in the payload will be metadata comments, and as such, clients can parse out metadata by reading comment lines until the first non-comment line is reached. Metadata will not be provided as columns.
    * - Units location
      - Express units only in the value column header via parentheses (e.g., ``value (cfs)``)
@@ -56,8 +56,8 @@ Key points
      - Kebab-case names
      - Keeps naming consistent with JSON and XML.
    * - Accept header for format and columns
-     - Use HTTP Accept header parameters to select date format and optional columns
-     - Default CSV serialization uses ISO-8601 strings. Examples: ``text/csv;date-format=ISO8601-Instant`` (default), ``text/csv;date-format=epoch-millis``. Use Accept header parameters to enable optional columns (e.g., ``quality=present``, ``data-entry-date=present``). If these were query params instead, toggling would be easier in a browser, but Accept keeps content negotiation consistent.
+     - Use query parameters to select date format and optional columns
+     - Default CSV serialization uses ISO-8601 strings. Examples: ``date-format=ISO8601-Instant`` (default), ``date-format=epoch-millis``. Likewise, optional columns can be toggled with ``include-optional-columns=true`` (default: false).
    * - Quality representation
      - ``quality`` (aka quality-code) is an optional integer bitmask
      - A bitmask (integer) compactly represents multiple boolean flags with fast native bitwise operations; a ``byte[]`` adds overhead without improving expressiveness for fixed flag sets.