docs: @Override
      public Map<String, FieldTypes> get(final Collection<String> fieldNames, Collection<String> indexNames) {
          // Shortcut - if we don't select any fields we don't have to do any database query
          if (fieldNames.isEmpty()) {
              return Collections.emptyMap();
          }
  
          // We have to transform the field type database entries to make them usable for the user.
          //
          // [
          //   {
          //     "index_name": "graylog_0",
          //     "fields": [
          //       {"field_name": "message", "physical_type": "text"},
          //       {"field_name": "source", "physical_type": "keyword"}
          //     ]
          //   },
          //   {
          //     "index_name": "graylog_1",
          //     "fields": [
          //       {"field_name": "message", "physical_type": "text"},
          //     ]
          //   }
          // ]
          //
          // gets transformed into
          //
          // {
          //   "message": {
          //     "field_name": "message",
          //     "types": [
          //       {
          //         "type": "string",
          //         "properties": ["full-text-search"],
          //         "index_names": ["graylog_0", "graylog_1"]
          //     ]
          //   },
          //   "source": {
          //     "field_name": "source",
          //     "types": [
          //       {
          //         "type": "string",
          //         "properties": ["enumerable"],
          //         "index_names": ["graylog_0"]
          //     ]
          //   }
          // }
  
          // field-name -> {physical-type -> [index-name, ...]}
          final Map<String, SetMultimap<String, String>> fields = new HashMap<>();
  
          // Convert the data from the database to be indexed by field name and physical type
          getTypesStream(fieldNames, indexNames).forEach(types -> {
              final String indexName = types.indexName();
  
              types.fields().forEach(fieldType -> {
                  final String fieldName = fieldType.fieldName();
                  final String physicalType = fieldType.physicalType();
  
                  if (fieldNames.contains(fieldName)) {
                      if (indexNames.isEmpty() || indexNames.contains(indexName)) {
                          if (!fields.containsKey(fieldName)) {
                              fields.put(fieldName, HashMultimap.create());
                          }
                          fields.get(fieldName).put(physicalType, indexName);
                      }
                  }
              });
          });
  
          final ImmutableMap.Builder<String, FieldTypes> result = ImmutableMap.builder();
  
          for (Map.Entry<String, SetMultimap<String, String>> fieldNameEntry : fields.entrySet()) {
              final String fieldName = fieldNameEntry.getKey();
              final Map<String, Collection<String>> physicalTypes = fieldNameEntry.getValue().asMap();
  
              // Use the field type mapper to do the conversion between the Elasticsearch type and our logical type
              final Set<FieldTypes.Type> types = physicalTypes.entrySet().stream()
                      .map(entry -> {
                          final String physicalType = entry.getKey();
                          final Set<String> indices = ImmutableSet.copyOf(entry.getValue());
  
                          return typeMapper.mapType(physicalType).map(t -> t.withIndexNames(indices));
                      })
                      .filter(Optional::isPresent)
                      .map(Optional::get)
                      .collect(Collectors.toSet());
  
              result.put(fieldName, FieldTypes.create(fieldName, types));
          }
  
          return result.build();
      }