code: private Map<String, InputColumn<?>> readSourceColumns(final SourceColumnMapping sourceColumnMapping,
              final AnalysisJobBuilder analysisJobBuilder, final SourceType source) {
          final Map<String, InputColumn<?>> inputColumns = new HashMap<>();
          final ColumnsType columnsType = source.getColumns();
          if (columnsType != null) {
              final List<ColumnType> columns = columnsType.getColumn();
              for (final ColumnType column : columns) {
                  final String path = column.getPath();
                  if (StringUtils.isNullOrEmpty(path)) {
                      throw new IllegalStateException("Column path cannot be null");
                  }
                  final Column physicalColumn = sourceColumnMapping.getColumn(path);
                  if (physicalColumn == null) {
                      logger.error("Column {} not found in {}", path, sourceColumnMapping);
                      throw new NoSuchColumnException(path);
                  }
  
                  final MetaModelInputColumn inputColumn = new MetaModelInputColumn(physicalColumn);
                  final String id = column.getId();
                  if (StringUtils.isNullOrEmpty(id)) {
                      throw new IllegalStateException("Source column id cannot be null");
                  }
  
                  final String expectedType = column.getType();
                  if (expectedType != null) {
                      final org.apache.metamodel.schema.ColumnType actualType = physicalColumn.getType();
                      if (actualType != null && !expectedType.equals(actualType.toString())) {
                          logger.warn("Column '{}' had type '{}', but '{}' was expected.",
                                  new Object[] { path, actualType, expectedType });
                      }
                  }
  
                  registerInputColumn(inputColumns, id, inputColumn);
                  analysisJobBuilder.addSourceColumn(inputColumn);
              }
          }
          return inputColumns;
      }
docs: Reads the source element of the job to extract a map of column IDs and
  related source {@link InputColumn}s.
  
  @param sourceColumnMapping
  @param analysisJobBuilder
  @param source
  @return