fix for write row

index.ts

@@ -107,7 +107,7 @@ export async function action(options: CSVRunOptions ) {
     for ( const entityChange of EntityChanges.parse(data).entityChanges ) {
       const writer = writers.get(entityChange.entity);
       const table = tables.get(entityChange.entity);
-      if ( !writer || !table ) throw new Error(`Table not found: ${entityChange.entity}`);
+      if ( !writer || !table ) continue; // skip if table not found
       const values = getValuesInEntityChange(entityChange);
       applyReservedFields(values, entityChange, cursor, clock);
 

package.json

@@ -1,5 +1,5 @@
 {
-    "version": "0.2.10",
+    "version": "0.2.11",
     "name": "substreams-sink-csv",
     "description": "Substreams Sink CSV",
     "type": "module",
@@ -24,7 +24,7 @@
         "prepublishOnly": "tsc"
     },
     "dependencies": {
-        "@substreams/sink-entity-changes": "^0.3.9",
+        "@substreams/sink-entity-changes": "^0.3.11",
         "log-update": "latest",
         "substreams-sink": "^0.16.0"
     },

schema.example.blobs.sql

@@ -0,0 +1,23 @@
+CREATE TABLE Blob
+(
+    id                              TEXT,
+    slot                            TEXT,
+    index                           TEXT,
+    blob                            TEXT,
+    kzg_commitment                  TEXT,
+    kzg_proof                       TEXT,
+    kzg_commitment_inclusion_proof  TEXT[],
+);
+
+CREATE TABLE Slot
+(
+    id                  TEXT,
+    number              BIGINT,
+    spec                TEXT,
+    proposer_index      INT,
+    parent_root         TEXT,
+    state_root          TEXT,
+    body_root           TEXT,
+    signature           TEXT,
+    timestamp           TEXT,
+);

src/getModuleHash.ts

@@ -17,7 +17,7 @@ export async function getModuleHash(options: CSVRunOptions) {
     for ( const module of substreamPackage.modules.modules ) {
       if ( module.name === options.moduleName ) {
         if ( !module.output ) throw new Error("Module has no output");
-        if ( module.output.type !== "proto:sf.substreams.sink.entity.v1.EntityChanges" ) throw new Error("Module output is not EntityChanges");
+        if ( !module.output.type.includes("entity.v1.EntityChanges") ) throw new Error("Module output is not EntityChanges");
         valid = true;
       }
     }

src/writeRow.spec.ts

@@ -3,12 +3,20 @@ import { formatValue } from "./writeRow.js";
 
 test("formatValue", () => {
     const options = {delimiter: ","};
-    // expect(formatValue("a", options)).toBe("a");
+    expect(formatValue("a", options)).toBe("a");
     expect(formatValue("'a'", options)).toBe("'a'");
     expect(formatValue("foo bar", options)).toBe("foo bar");
     expect(formatValue("foo \" bar", options)).toBe("foo \"\" bar");
     expect(formatValue(undefined, options)).toBe("");
     expect(formatValue(null, options)).toBe("");
+    expect(formatValue("0", options)).toBe("0");
     expect(formatValue("a,b", options)).toBe("\"a,b\"");
     expect(formatValue("a,\"b", options)).toBe("\"a,\"\"b\"");
+
+    // Array
+    expect(formatValue(["ab", "cd"], options)).toBe("\"[\"\"ab\"\",\"\"cd\"\"]\"");
 })
+
+test.skip("formatValue - recursive array", () => {
+    // https://github.com/substreams-js/substreams-sink-entity-changes/issues/5
+});

src/writeRow.ts

@@ -9,9 +9,13 @@ export function writeRow(writer: fs.WriteStream, columns: any[], options: WriteR
     writer.write(columns.join(options.delimiter) + '\n');
 }
 
-export function formatValue(value: string|undefined|null, options: WriteRowOptions): string {
+export function formatValue(value: string|undefined|null|object, options: WriteRowOptions): string {
     if (value === undefined || value === null) return "";
 
+    if (typeof value == "object") {
+        value = JSON.stringify(value);
+    }
+
     if (typeof value == "string") {
         value = value.replace(/"/g, "\"\"")
         if ( value.includes(options.delimiter) ) value = `"${value}"`; // escape commas

version.ts

@@ -1 +1 @@
-export const version = "0.2.10";
+export const version = "0.2.11";