PhoenixPigSchemaUtil { public static ResourceSchema getResourceSchema(final Configuration configuration) throws IOException { final ResourceSchema schema = new ResourceSchema(); try { List<ColumnInfo> columns = null; final SchemaType schemaType = PhoenixConfigurationUtil.getSchemaType(configuration); if(SchemaType.QUERY.equals(schemaType)) { final String sqlQuery = PhoenixConfigurationUtil.getSelectStatement(configuration); Preconditions.checkNotNull(sqlQuery, "No Sql Query exists within the configuration"); final SqlQueryToColumnInfoFunction function = new SqlQueryToColumnInfoFunction(configuration); columns = function.apply(sqlQuery); } else { columns = PhoenixConfigurationUtil.getSelectColumnMetadataList(configuration); } ResourceFieldSchema fields[] = new ResourceFieldSchema[columns.size()]; int i = 0; for(ColumnInfo cinfo : columns) { int sqlType = cinfo.getSqlType(); PDataType phoenixDataType = PDataType.fromTypeId(sqlType); byte pigType = TypeUtil.getPigDataTypeForPhoenixType(phoenixDataType); ResourceFieldSchema field = new ResourceFieldSchema(); field.setType(pigType).setName(cinfo.getDisplayName()); fields[i++] = field; } schema.setFields(fields); } catch(SQLException sqle) { LOG.error(String.format("Error: SQLException [%s] ",sqle.getMessage())); throw new IOException(sqle); } return schema; } private  PhoenixPigSchemaUtil(); static ResourceSchema getResourceSchema(final Configuration configuration);  }
@Test public void testSchema() throws SQLException, IOException { final Configuration configuration = mock(Configuration.class); final List<ColumnInfo> columnInfos = ImmutableList.of(ID_COLUMN,NAME_COLUMN); final String encodedColumnInfos = ColumnInfoToStringEncoderDecoder.encode(columnInfos); when(configuration.get(PhoenixConfigurationUtil.SELECT_COLUMN_INFO_KEY)).thenReturn(encodedColumnInfos); when(configuration.get(PhoenixConfigurationUtil.SCHEMA_TYPE)).thenReturn(SchemaType.TABLE.name()); final ResourceSchema actual = PhoenixPigSchemaUtil.getResourceSchema(configuration); final ResourceFieldSchema[] fields = new ResourceFieldSchema[2]; fields[0] = new ResourceFieldSchema().setName("ID") .setType(DataType.LONG); fields[1] = new ResourceFieldSchema().setName("NAME") .setType(DataType.CHARARRAY); final ResourceSchema expected = new ResourceSchema().setFields(fields); assertEquals(expected.toString(), actual.toString()); }
@Test(expected=IllegalDataException.class) public void testUnSupportedTypes() throws SQLException, IOException { final Configuration configuration = mock(Configuration.class); final List<ColumnInfo> columnInfos = ImmutableList.of(ID_COLUMN,LOCATION_COLUMN); final String encodedColumnInfos = ColumnInfoToStringEncoderDecoder.encode(columnInfos); when(configuration.get(PhoenixConfigurationUtil.SELECT_COLUMN_INFO_KEY)).thenReturn(encodedColumnInfos); PhoenixPigSchemaUtil.getResourceSchema(configuration); fail("We currently don't support Array type yet. WIP!!"); }