public void processRow(GenericRow decodedRow, Result reusedResult)
      throws Exception {
    reusedResult.reset();
    if (_complexTypeTransformer != null) {
      // TODO: consolidate complex type transformer into composite type transformer
      decodedRow = _complexTypeTransformer.transform(decodedRow);
    }
    Collection<GenericRow> rows = (Collection<GenericRow>) decodedRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY);
    if (rows != null) {
      for (GenericRow row : rows) {
        processPlainRow(row, reusedResult);
      }
    } else {
      processPlainRow(decodedRow, reusedResult);
    }
  }
@Test
  public void testSingleRow()
      throws Exception {
    TableConfig config = createTestTableConfig();
    Schema schema = Fixtures.createSchema();
    TransformPipeline pipeline = new TransformPipeline(config, schema);
    GenericRow simpleRow = Fixtures.createSingleRow(9527);
    TransformPipeline.Result result = new TransformPipeline.Result();
    pipeline.processRow(simpleRow, result);
    Assert.assertNotNull(result);
    Assert.assertEquals(result.getTransformedRows().size(), 1);
    Assert.assertEquals(result.getSkippedRowCount(), 0);
    Assert.assertEquals(result.getTransformedRows().get(0), simpleRow);
  }