OutputColsHelper implements Serializable { public TableSchema getResultSchema() { int resultLength = reservedCols.length + outputColNames.length; String[] resultColNames = new String[resultLength]; TypeInformation[] resultColTypes = new TypeInformation[resultLength]; for (int i = 0; i < reservedCols.length; i++) { resultColNames[reservedColsPosInResult[i]] = inputColNames[reservedCols[i]]; resultColTypes[reservedColsPosInResult[i]] = inputColTypes[reservedCols[i]]; } for (int i = 0; i < outputColsPosInResult.length; i++) { resultColNames[outputColsPosInResult[i]] = outputColNames[i]; resultColTypes[outputColsPosInResult[i]] = outputColTypes[i]; } return new TableSchema(resultColNames, resultColTypes); } OutputColsHelper(TableSchema inputSchema, String outputColName, TypeInformation outputColType);  OutputColsHelper(TableSchema inputSchema, String outputColName, TypeInformation outputColType,
                            String[] reservedColNames);  OutputColsHelper(TableSchema inputSchema, String[] outputColNames, TypeInformation[] outputColTypes);  OutputColsHelper(TableSchema inputSchema, String[] outputColNames, TypeInformation[] outputColTypes,
                            String[] reservedColNames); String[] getReservedColumns(); TableSchema getResultSchema(); Row getResultRow(Row input, Row output);  }
@Test public void testResultSchema() { TableSchema expectSchema = new TableSchema( new String[]{"f0", "f1", "f2", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Long.class), TypeInformation.of(Integer.class), TypeInformation.of(String.class) } ); OutputColsHelper helper = new OutputColsHelper( tableSchema, "res", TypeInformation.of(String.class) ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(String.class) } ); helper = new OutputColsHelper( tableSchema, "res", TypeInformation.of(String.class), reservedColNames ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "res1", "res2"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(String.class), TypeInformation.of(Integer.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"res1", "res2"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Integer.class)}, reservedColNames ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "f2", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Long.class), TypeInformation.of(Integer.class), TypeInformation.of(String.class) } ); helper = new OutputColsHelper( tableSchema, "res", TypeInformation.of(String.class) ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "f2"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(Long.class), TypeInformation.of(Integer.class) } ); helper = new OutputColsHelper( tableSchema, "f0", TypeInformation.of(Integer.class) ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "f2"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(Long.class), TypeInformation.of(String.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"f0", "f2"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class) } ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Integer.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"res", "f0"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class) }, reservedColNames ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Long.class), TypeInformation.of(Integer.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"res"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class)}, new String[]{"f1", "f0"} ); Assert.assertEquals(helper.getResultSchema(), expectSchema); }
@Test public void testExceptionCase() { TableSchema expectSchema = new TableSchema( new String[]{"f0", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Integer.class) } ); OutputColsHelper helper = new OutputColsHelper( tableSchema, new String[]{"res", "f0"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class) }, new String[]{"res", "res2"} ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "res"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Long.class), TypeInformation.of(Integer.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"res", "f0"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class) }, new String[]{"f1", "res"} ); Assert.assertEquals(helper.getResultSchema(), expectSchema); expectSchema = new TableSchema( new String[]{"f0", "f1", "f2"}, new TypeInformation[]{ TypeInformation.of(String.class), TypeInformation.of(Integer.class), TypeInformation.of(Double.class) } ); helper = new OutputColsHelper( tableSchema, new String[]{"f1", "f0", "f2"}, new TypeInformation[]{ TypeInformation.of(Integer.class), TypeInformation.of(String.class), TypeInformation.of(Double.class) }, new String[]{"f1", "res"} ); Assert.assertEquals(helper.getResultSchema(), expectSchema); }