SamzaExecutor implements SqlExecutor { @Override public SqlSchema getTableSchema(ExecutionContext context, String tableName) throws ExecutorException { int execId = execIdSeq.incrementAndGet(); Map<String, String> staticConfigs = fetchSamzaSqlConfig(execId); Config samzaSqlConfig = new MapConfig(staticConfigs); SqlSchema sqlSchema; try { SqlIOResolver ioResolver = SamzaSqlApplicationConfig.createIOResolver(samzaSqlConfig); SqlIOConfig sourceInfo = ioResolver.fetchSourceInfo(tableName); RelSchemaProvider schemaProvider = SamzaSqlApplicationConfig.initializePlugin("RelSchemaProvider", sourceInfo.getRelSchemaProviderName(), samzaSqlConfig, SamzaSqlApplicationConfig.CFG_FMT_REL_SCHEMA_PROVIDER_DOMAIN, (o, c) -> ((RelSchemaProviderFactory) o).create(sourceInfo.getSystemStream(), c)); sqlSchema = schemaProvider.getSqlSchema(); } catch (SamzaException ex) { throw new ExecutorException(ex); } return sqlSchema; }  @Override void start(ExecutionContext context); @Override void stop(ExecutionContext context); @Override EnvironmentVariableHandler getEnvironmentVariableHandler(); @Override List<String> listTables(ExecutionContext context); @Override SqlSchema getTableSchema(ExecutionContext context, String tableName); @Override QueryResult executeQuery(ExecutionContext context, String statement); @Override int getRowCount(); @Override List<String[]> retrieveQueryResult(ExecutionContext context, int startRow, int endRow); @Override List<String[]> consumeQueryResult(ExecutionContext context, int startRow, int endRow); @Override NonQueryResult executeNonQuery(ExecutionContext context, File sqlFile); @Override NonQueryResult executeNonQuery(ExecutionContext context, List<String> statement); @Override void stopExecution(ExecutionContext context, int exeId); @Override void removeExecution(ExecutionContext context, int exeId); @Override ExecutionStatus queryExecutionStatus(int execId); @Override List<SqlFunction> listFunctions(ExecutionContext context); @Override String getVersion();  }
@Test public void testGetTableSchema() throws ExecutorException { prepareEnvironmentVariable(); SqlSchema ts = mExecutor.getTableSchema(new ExecutionContext(), "kafka.ProfileChangeStream"); List<SqlSchema.SqlField> fields = ts.getFields(); Assert.assertEquals("Name", fields.get(0).getFieldName()); Assert.assertEquals("NewCompany", fields.get(1).getFieldName()); Assert.assertEquals("OldCompany", fields.get(2).getFieldName()); Assert.assertEquals("ProfileChangeTimestamp", fields.get(3).getFieldName()); Assert.assertEquals("STRING", fields.get(0).getFieldSchema().getFieldType().toString()); Assert.assertEquals("STRING", fields.get(1).getFieldSchema().getFieldType().toString()); Assert.assertEquals("STRING", fields.get(2).getFieldSchema().getFieldType().toString()); Assert.assertEquals("INT64", fields.get(3).getFieldSchema().getFieldType().toString()); }