Feature - collect and report multiple field errors (#75)

Problem
=======
This PR is intended to implement 2 enhancements to schema error
reporting.
* When a parquet schema includes an invalid type, encoding or
compression the current error does not indicate which column has the the
problem
* When a parquet schema has multiple issues, the code currently fails on
the first, making multiple errors quite cumbersome

Solution
========
Modified the schema.ts and added tests to:
* Change error messages from the original `invalid parquet type:
UNKNOWN` to `invalid parquet type: UNKNOWN, for Column: quantity`
* Keep track of schema errors as we loop through each column in the
schema, and at the end, if there are any errors report them all as
below:
`invalid parquet type: UNKNOWN, for Column: quantity`
`invalid parquet type: UNKNOWN, for Column: value`

Change summary:
---------------
* adding tests and code to ensure multiple field errors are logged, as
well as indicating which column had the error
* also adding code to handle multiple encoding and compression schema
issues

Steps to Verify:
----------------
1. Download this [parquet
file](https://usaz02prismdevmlaas01.blob.core.windows.net/ml-job-config/dataSets/multiple-unsupported-columns.parquet?sv=2020-10-02&st=2023-01-09T15%3A28%3A09Z&se=2025-01-10T15%3A28%3A00Z&sr=b&sp=r&sig=GS0Skk93DCn5CnC64DbnIH2U7JhzHM2nnhq1U%2B2HwPs%3D)
2. attempt to open this parquet with this library `const reader = await
parquet.ParquetReader.openFile(<path to parquet file>)`
3. You should receive errors for more than one column, which also
includes the column name for each error

---------

Co-authored-by: Wil Wade <[email protected]>

Author: dgaudet

lib/schema.ts

@@ -81,6 +81,7 @@ function buildFields(schema: SchemaDefinition, rLevelParentMax?: number, dLevelP
   }
 
   let fieldList: Record<string, ParquetField> = {};
+  let fieldErrors: Array<string> = [];
   for (let name in schema) {
     const opts = schema[name];
 
@@ -129,9 +130,15 @@ function buildFields(schema: SchemaDefinition, rLevelParentMax?: number, dLevelP
       continue;
     }
 
+    let nameWithPath = (`${name}` || 'missing name')
+    if (path && path.length > 0) {
+      nameWithPath = `${path}.${nameWithPath}`
+    }
+
     const typeDef = opts.type ? parquet_types.PARQUET_LOGICAL_TYPES[opts.type] : undefined;
     if (!typeDef) {
-      throw 'invalid parquet type: ' + (opts.type || "missing type");
+      fieldErrors.push(`Invalid parquet type: ${(opts.type || "missing type")}, for Column: ${nameWithPath}`);
+      continue;
     }
 
     /* field encoding */
@@ -140,15 +147,15 @@ function buildFields(schema: SchemaDefinition, rLevelParentMax?: number, dLevelP
     }
 
     if (!(opts.encoding in parquet_codec)) {
-      throw 'unsupported parquet encoding: ' + opts.encoding;
+      fieldErrors.push(`Unsupported parquet encoding: ${opts.encoding}, for Column: ${nameWithPath}`);
     }
 
     if (!opts.compression) {
       opts.compression = 'UNCOMPRESSED';
     }
 
     if (!(opts.compression in parquet_compression.PARQUET_COMPRESSION_METHODS)) {
-      throw 'unsupported compression method: ' + opts.compression;
+      fieldErrors.push(`Unsupported compression method: ${opts.compression}, for Column: ${nameWithPath}`);
     }
 
     /* add to schema */
@@ -167,6 +174,10 @@ function buildFields(schema: SchemaDefinition, rLevelParentMax?: number, dLevelP
     };
   }
 
+  if (fieldErrors.length > 0) {
+    throw fieldErrors.reduce((accumulator, currentVal) => accumulator + '\n' + currentVal);
+  }
+
   return fieldList;
 }
 

test/schema.js

@@ -467,4 +467,61 @@ describe('ParquetSchema', function() {
     }
   });
 
+  it('should indicate which column had an invalid type in a simple flat schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        quantity: {type: 'UNKNOWN'},
+      })
+    }, 'Invalid parquet type: UNKNOWN, for Column: quantity');
+  });
+
+  it('should indicate each column which has an invalid type in a simple flat schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        quantity: {type: 'UNKNOWN'},
+        value: {type: 'UNKNOWN'},
+      })
+    }, 'Invalid parquet type: UNKNOWN, for Column: quantity\nInvalid parquet type: UNKNOWN, for Column: value');
+  });
+
+  it('should indicate each column which has an invalid type when one is correct in a simple flat schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        quantity: {type: 'INT32'},
+        value: {type: 'UNKNOWN'},
+      })
+    }, 'Invalid parquet type: UNKNOWN, for Column: value');
+  });
+
+  it('should indicate each column which has an invalid type in a nested schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        name: { type: 'UTF8' },
+        stock: {
+          fields: {
+            quantity: { type: 'UNKNOWN' },
+            warehouse: { type: 'UNKNOWN' },
+          }
+        },
+        price: { type: 'UNKNOWN' },
+      })
+    }, 'Invalid parquet type: UNKNOWN, for Column: stock.quantity\nInvalid parquet type: UNKNOWN, for Column: stock.warehouse');
+  });
+
+  it('should indicate which column had an invalid encoding in a simple flat schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        quantity: {type: 'INT32', encoding: 'UNKNOWN'},
+      })
+    }, 'Unsupported parquet encoding: UNKNOWN, for Column: quantity');
+  });
+
+  it('should indicate which column had an invalid compression type in a simple flat schema', function() {
+    assert.throws(() => {
+      new parquet.ParquetSchema({
+        quantity: {type: 'INT32', compression: 'UNKNOWN'},
+      })
+    }, 'Unsupported compression method: UNKNOWN, for Column: quantity');
+  });
+
 });