Skip to content

feat: web streams based encoding/csv #1993

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Apr 6, 2022
Merged

feat: web streams based encoding/csv #1993

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c34f1b7
feat: web streams based encoding/csv
uki00a Mar 3, 2022
c573f87
fix: Rewrite CSVStream not to depend on Deno.Reader
uki00a Mar 5, 2022
6d437d6
Merge branch 'main' into csv_stream
uki00a Mar 5, 2022
e197808
fix: added test case for error handling and fixed bug
uki00a Mar 7, 2022
ebebc80
Merge branch 'main' into csv_stream
uki00a Apr 3, 2022
31bec60
fix(encoding/csv/stream): Make `CSVStream` be a `TransformStream<stri…
uki00a Apr 3, 2022
2723e79
fix(encoding/csv): Make `ParseError` extend `SyntaxError` instaed of …
uki00a Apr 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
288 changes: 48 additions & 240 deletions encoding/csv.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,22 @@ import { BufReader } from "../io/buffer.ts";
import { TextProtoReader } from "../textproto/mod.ts";
import { StringReader } from "../io/readers.ts";
import { assert } from "../_util/assert.ts";
import {
ERR_FIELD_COUNT,
ERR_INVALID_DELIM,
ParseError,
readRecord,
} from "./csv/_io.ts";
import type { LineReader, ReadOptions } from "./csv/_io.ts";

export {
ERR_BARE_QUOTE,
ERR_FIELD_COUNT,
ERR_INVALID_DELIM,
ERR_QUOTE,
ParseError,
} from "./csv/_io.ts";
export type { ReadOptions } from "./csv/_io.ts";
export { NEWLINE, stringify, StringifyError } from "./csv_stringify.ts";

export type {
Expand All @@ -18,66 +33,45 @@ export type {
StringifyOptions,
} from "./csv_stringify.ts";

const INVALID_RUNE = ["\r", "\n", '"'];
class TextProtoLineReader implements LineReader {
#tp: TextProtoReader;
constructor(bufReader: BufReader) {
this.#tp = new TextProtoReader(bufReader);
}

export const ERR_BARE_QUOTE = 'bare " in non-quoted-field';
export const ERR_QUOTE = 'extraneous or missing " in quoted-field';
export const ERR_INVALID_DELIM = "Invalid Delimiter";
export const ERR_FIELD_COUNT = "wrong number of fields";
async readLine() {
let line: string;
const r = await this.#tp.readLine();
if (r === null) return null;
line = r;

/**
* A ParseError is returned for parsing errors.
* Line numbers are 1-indexed and columns are 0-indexed.
*/
export class ParseError extends Error {
/** Line where the record starts*/
startLine: number;
/** Line where the error occurred */
line: number;
/** Column (rune index) where the error occurred */
column: number | null;

constructor(
start: number,
line: number,
column: number | null,
message: string,
) {
super();
this.startLine = start;
this.column = column;
this.line = line;
// For backwards compatibility, drop trailing \r before EOF.
if (
(await this.isEOF()) && line.length > 0 && line[line.length - 1] === "\r"
) {
line = line.substring(0, line.length - 1);
}

if (message === ERR_FIELD_COUNT) {
this.message = `record on line ${line}: ${message}`;
} else if (start !== line) {
this.message =
`record on line ${start}; parse error on line ${line}, column ${column}: ${message}`;
} else {
this.message =
`parse error on line ${line}, column ${column}: ${message}`;
// Normalize \r\n to \n on all input lines.
if (
line.length >= 2 &&
line[line.length - 2] === "\r" &&
line[line.length - 1] === "\n"
) {
line = line.substring(0, line.length - 2);
line = line + "\n";
}

return line;
}
}

/**
* @property separator - Character which separates values. Default: ','
* @property comment - Character to start a comment. Default: '#'
* @property trimLeadingSpace - Flag to trim the leading space of the value.
* Default: 'false'
* @property lazyQuotes - Allow unquoted quote in a quoted field or non double
* quoted quotes in quoted field. Default: 'false'
* @property fieldsPerRecord - Enabling the check of fields for each row.
* If == 0, first row is used as referral for the number of fields.
*/
export interface ReadOptions {
separator?: string;
comment?: string;
trimLeadingSpace?: boolean;
lazyQuotes?: boolean;
fieldsPerRecord?: number;
async isEOF() {
return (await this.#tp.r.peek(0)) === null;
}
}

const INVALID_RUNE = ["\r", "\n", '"'];

function chkOptions(opt: ReadOptions): void {
if (!opt.separator) {
opt.separator = ",";
Expand All @@ -94,193 +88,6 @@ function chkOptions(opt: ReadOptions): void {
}
}

async function readRecord(
startLine: number,
reader: BufReader,
opt: ReadOptions = { separator: ",", trimLeadingSpace: false },
): Promise<string[] | null> {
const tp = new TextProtoReader(reader);
let line = await readLine(tp);
let lineIndex = startLine + 1;

if (line === null) return null;
if (line.length === 0) {
return [];
}
// line starting with comment character is ignored
if (opt.comment && line[0] === opt.comment) {
return [];
}

assert(opt.separator != null);

let fullLine = line;
let quoteError: ParseError | null = null;
const quote = '"';
const quoteLen = quote.length;
const separatorLen = opt.separator.length;
let recordBuffer = "";
const fieldIndexes = [] as number[];
parseField:
for (;;) {
if (opt.trimLeadingSpace) {
line = line.trimLeft();
}

if (line.length === 0 || !line.startsWith(quote)) {
// Non-quoted string field
const i = line.indexOf(opt.separator);
let field = line;
if (i >= 0) {
field = field.substring(0, i);
}
// Check to make sure a quote does not appear in field.
if (!opt.lazyQuotes) {
const j = field.indexOf(quote);
if (j >= 0) {
const col = runeCount(
fullLine.slice(0, fullLine.length - line.slice(j).length),
);
quoteError = new ParseError(
startLine + 1,
lineIndex,
col,
ERR_BARE_QUOTE,
);
break parseField;
}
}
recordBuffer += field;
fieldIndexes.push(recordBuffer.length);
if (i >= 0) {
line = line.substring(i + separatorLen);
continue parseField;
}
break parseField;
} else {
// Quoted string field
line = line.substring(quoteLen);
for (;;) {
const i = line.indexOf(quote);
if (i >= 0) {
// Hit next quote.
recordBuffer += line.substring(0, i);
line = line.substring(i + quoteLen);
if (line.startsWith(quote)) {
// `""` sequence (append quote).
recordBuffer += quote;
line = line.substring(quoteLen);
} else if (line.startsWith(opt.separator)) {
// `","` sequence (end of field).
line = line.substring(separatorLen);
fieldIndexes.push(recordBuffer.length);
continue parseField;
} else if (0 === line.length) {
// `"\n` sequence (end of line).
fieldIndexes.push(recordBuffer.length);
break parseField;
} else if (opt.lazyQuotes) {
// `"` sequence (bare quote).
recordBuffer += quote;
} else {
// `"*` sequence (invalid non-escaped quote).
const col = runeCount(
fullLine.slice(0, fullLine.length - line.length - quoteLen),
);
quoteError = new ParseError(
startLine + 1,
lineIndex,
col,
ERR_QUOTE,
);
break parseField;
}
} else if (line.length > 0 || !(await isEOF(tp))) {
// Hit end of line (copy all data so far).
recordBuffer += line;
const r = await readLine(tp);
lineIndex++;
line = r ?? ""; // This is a workaround for making this module behave similarly to the encoding/csv/reader.go.
fullLine = line;
if (r === null) {
// Abrupt end of file (EOF or error).
if (!opt.lazyQuotes) {
const col = runeCount(fullLine);
quoteError = new ParseError(
startLine + 1,
lineIndex,
col,
ERR_QUOTE,
);
break parseField;
}
fieldIndexes.push(recordBuffer.length);
break parseField;
}
recordBuffer += "\n"; // preserve line feed (This is because TextProtoReader removes it.)
} else {
// Abrupt end of file (EOF on error).
if (!opt.lazyQuotes) {
const col = runeCount(fullLine);
quoteError = new ParseError(
startLine + 1,
lineIndex,
col,
ERR_QUOTE,
);
break parseField;
}
fieldIndexes.push(recordBuffer.length);
break parseField;
}
}
}
}
if (quoteError) {
throw quoteError;
}
const result = [] as string[];
let preIdx = 0;
for (const i of fieldIndexes) {
result.push(recordBuffer.slice(preIdx, i));
preIdx = i;
}
return result;
}

async function isEOF(tp: TextProtoReader): Promise<boolean> {
return (await tp.r.peek(0)) === null;
}

function runeCount(s: string): number {
// Array.from considers the surrogate pair.
return Array.from(s).length;
}

async function readLine(tp: TextProtoReader): Promise<string | null> {
let line: string;
const r = await tp.readLine();
if (r === null) return null;
line = r;

// For backwards compatibility, drop trailing \r before EOF.
if ((await isEOF(tp)) && line.length > 0 && line[line.length - 1] === "\r") {
line = line.substring(0, line.length - 1);
}

// Normalize \r\n to \n on all input lines.
if (
line.length >= 2 &&
line[line.length - 2] === "\r" &&
line[line.length - 1] === "\n"
) {
line = line.substring(0, line.length - 2);
line = line + "\n";
}

return line;
}

/**
* Parse the CSV from the `reader` with the options provided and return `string[][]`.
*
Expand All @@ -302,8 +109,9 @@ export async function readMatrix(
let lineIndex = 0;
chkOptions(opt);

const lineReader = new TextProtoLineReader(reader);
for (;;) {
const r = await readRecord(lineIndex, reader, opt);
const r = await readRecord(lineIndex, lineReader, opt);
if (r === null) break;
lineResult = r;
lineIndex++;
Expand Down
Loading