path: root/icing/schema/schema-util.cc

// Copyright (C) 2019 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "icing/schema/schema-util.h"

#include <cstdint>
#include <string>
#include <string_view>
#include <unordered_map>
#include <unordered_set>
#include <utility>

#include "icing/text_classifier/lib3/utils/base/status.h"
#include "icing/absl_ports/annotate.h"
#include "icing/absl_ports/canonical_errors.h"
#include "icing/absl_ports/str_cat.h"
#include "icing/absl_ports/str_join.h"
#include "icing/legacy/core/icing-string-util.h"
#include "icing/proto/schema.pb.h"
#include "icing/proto/term.pb.h"
#include "icing/util/logging.h"
#include "icing/util/status-macros.h"

namespace icing {
namespace lib {

namespace {

bool ArePropertiesEqual(const PropertyConfigProto& old_property,
                        const PropertyConfigProto& new_property) {
  return old_property.property_name() == new_property.property_name() &&
         old_property.data_type() == new_property.data_type() &&
         old_property.schema_type() == new_property.schema_type() &&
         old_property.cardinality() == new_property.cardinality() &&
         old_property.string_indexing_config().term_match_type() ==
             new_property.string_indexing_config().term_match_type() &&
         old_property.string_indexing_config().tokenizer_type() ==
             new_property.string_indexing_config().tokenizer_type() &&
         old_property.document_indexing_config().index_nested_properties() ==
             new_property.document_indexing_config().index_nested_properties();
}

bool IsIndexedProperty(const PropertyConfigProto& property_config) {
  switch (property_config.data_type()) {
    case PropertyConfigProto::DataType::STRING:
      return property_config.string_indexing_config().term_match_type() !=
             TermMatchType::UNKNOWN;
    case PropertyConfigProto::DataType::INT64:
      return property_config.integer_indexing_config().numeric_match_type() !=
             IntegerIndexingConfig::NumericMatchType::UNKNOWN;
    case PropertyConfigProto::DataType::UNKNOWN:
    case PropertyConfigProto::DataType::DOUBLE:
    case PropertyConfigProto::DataType::BOOLEAN:
    case PropertyConfigProto::DataType::BYTES:
    case PropertyConfigProto::DataType::DOCUMENT:
      return false;
  }
}

bool IsCardinalityCompatible(const PropertyConfigProto& old_property,
                             const PropertyConfigProto& new_property) {
  if (old_property.cardinality() < new_property.cardinality()) {
    // We allow a new, less restrictive cardinality (i.e. a REQUIRED field
    // can become REPEATED or OPTIONAL, but not the other way around).
    ICING_VLOG(1) << absl_ports::StrCat(
        "Cardinality is more restrictive than before ",
        PropertyConfigProto::Cardinality::Code_Name(old_property.cardinality()),
        "->",
        PropertyConfigProto::Cardinality::Code_Name(
            new_property.cardinality()));
    return false;
  }
  return true;
}

bool IsDataTypeCompatible(const PropertyConfigProto& old_property,
                          const PropertyConfigProto& new_property) {
  if (old_property.data_type() != new_property.data_type()) {
    // TODO(cassiewang): Maybe we can be a bit looser with this, e.g. we just
    // string cast an int64_t to a string. But for now, we'll stick with
    // simplistics.
    ICING_VLOG(1) << absl_ports::StrCat(
        "Data type ",
        PropertyConfigProto::DataType::Code_Name(old_property.data_type()),
        "->",
        PropertyConfigProto::DataType::Code_Name(new_property.data_type()));
    return false;
  }
  return true;
}

bool IsSchemaTypeCompatible(const PropertyConfigProto& old_property,
                            const PropertyConfigProto& new_property) {
  if (old_property.schema_type() != new_property.schema_type()) {
    ICING_VLOG(1) << absl_ports::StrCat("Schema type ",
                                        old_property.schema_type(), "->",
                                        new_property.schema_type());
    return false;
  }
  return true;
}

bool IsPropertyCompatible(const PropertyConfigProto& old_property,
                          const PropertyConfigProto& new_property) {
  return IsDataTypeCompatible(old_property, new_property) &&
         IsSchemaTypeCompatible(old_property, new_property) &&
         IsCardinalityCompatible(old_property, new_property);
}

bool IsTermMatchTypeCompatible(const StringIndexingConfig& old_indexed,
                               const StringIndexingConfig& new_indexed) {
  return old_indexed.term_match_type() == new_indexed.term_match_type() &&
         old_indexed.tokenizer_type() == new_indexed.tokenizer_type();
}

bool IsIntegerNumericMatchTypeCompatible(
    const IntegerIndexingConfig& old_indexed,
    const IntegerIndexingConfig& new_indexed) {
  return old_indexed.numeric_match_type() == new_indexed.numeric_match_type();
}

void AddIncompatibleChangeToDelta(
    std::unordered_set<std::string>& incompatible_delta,
    const SchemaTypeConfigProto& old_type_config,
    const SchemaUtil::DependentMap& new_schema_dependent_map,
    const SchemaUtil::TypeConfigMap& old_type_config_map,
    const SchemaUtil::TypeConfigMap& new_type_config_map) {
  // If this type is incompatible, then every type that depends on it might
  // also be incompatible. Use the dependent map to mark those ones as
  // incompatible too.
  incompatible_delta.insert(old_type_config.schema_type());
  auto dependent_types_itr =
      new_schema_dependent_map.find(old_type_config.schema_type());
  if (dependent_types_itr != new_schema_dependent_map.end()) {
    for (std::string_view dependent_type : dependent_types_itr->second) {
      // The types from new_schema that depend on the current
      // old_type_config may not present in old_schema.
      // Those types will be listed at schema_delta.schema_types_new
      // instead.
      std::string dependent_type_str(dependent_type);
      if (old_type_config_map.find(dependent_type_str) !=
          old_type_config_map.end()) {
        incompatible_delta.insert(std::move(dependent_type_str));
      }
    }
  }
}

}  // namespace

libtextclassifier3::Status ExpandTranstiveDependents(
    const SchemaUtil::DependentMap& dependent_map, std::string_view type,
    SchemaUtil::DependentMap* expanded_dependent_map,
    std::unordered_set<std::string_view>* pending_expansions,
    std::unordered_set<std::string_view>* orphaned_types) {
  auto expanded_itr = expanded_dependent_map->find(type);
  if (expanded_itr != expanded_dependent_map->end()) {
    // We've already expanded this type. Just return.
    return libtextclassifier3::Status::OK;
  }
  auto itr = dependent_map.find(type);
  if (itr == dependent_map.end()) {
    // It's an orphan. Just return.
    orphaned_types->insert(type);
    return libtextclassifier3::Status::OK;
  }
  pending_expansions->insert(type);
  std::unordered_set<std::string_view> expanded_dependents;

  // Add all of the direct dependents.
  expanded_dependents.reserve(itr->second.size());
  expanded_dependents.insert(itr->second.begin(), itr->second.end());

  // Iterate through each direct dependent and add their indirect dependents.
  for (std::string_view dep : itr->second) {
    // 1. Check if we're in the middle of expanding this type - IOW there's a
    // cycle!
    if (pending_expansions->count(dep) > 0) {
      return absl_ports::InvalidArgumentError(
          absl_ports::StrCat("Infinite loop detected in type configs. '", type,
                             "' references itself."));
    }

    // 2. Expand this type as needed.
    ICING_RETURN_IF_ERROR(
        ExpandTranstiveDependents(dependent_map, dep, expanded_dependent_map,
                                  pending_expansions, orphaned_types));
    if (orphaned_types->count(dep) > 0) {
      // Dep is an orphan. Just skip to the next dep.
      continue;
    }

    // 3. Dep has been fully expanded. Add all of its dependents to this
    // type's dependents.
    auto dep_expanded_itr = expanded_dependent_map->find(dep);
    expanded_dependents.reserve(expanded_dependents.size() +
                                dep_expanded_itr->second.size());
    expanded_dependents.insert(dep_expanded_itr->second.begin(),
                               dep_expanded_itr->second.end());
  }
  expanded_dependent_map->insert({type, std::move(expanded_dependents)});
  pending_expansions->erase(type);
  return libtextclassifier3::Status::OK;
}

// Calculate and return the transitive closure of dependent_map, which expands
// the dependent_map to also include indirect dependents
//
// Ex. Suppose we have a schema with three types A, B and C, and we have the
// following dependent relationship.
//
// C -> B (B depends on C)
// B -> A (A depends on B)
//
// Then, this function would expand the map by adding C -> A to the map.
libtextclassifier3::StatusOr<SchemaUtil::DependentMap>
ExpandTranstiveDependents(const SchemaUtil::DependentMap& dependent_map) {
  SchemaUtil::DependentMap expanded_dependent_map;

  // Types that we are expanding.
  std::unordered_set<std::string_view> pending_expansions;

  // Types that have no dependents.
  std::unordered_set<std::string_view> orphaned_types;
  for (const auto& kvp : dependent_map) {
    ICING_RETURN_IF_ERROR(ExpandTranstiveDependents(
        dependent_map, kvp.first, &expanded_dependent_map, &pending_expansions,
        &orphaned_types));
  }
  return expanded_dependent_map;
}

// Builds a transitive dependent map. 'Orphaned' types (types with no
// dependents) will not be present in the map.
//
// Ex. Suppose we have a schema with four types A, B, C, D. A has a property of
// type B and B has a property of type C. C and D only have non-document
// properties.
//
// The transitive dependent map for this schema would be:
// C -> A, B (both A and B depend on C)
// B -> A (A depends on B)
//
// A and D would be considered orphaned properties because no type refers to
// them.
//
// RETURNS:
//   On success, a transitive dependent map of all types in the schema.
//   INVALID_ARGUMENT if the schema contains a cycle or an undefined type.
//   ALREADY_EXISTS if a schema type is specified more than once in the schema
libtextclassifier3::StatusOr<SchemaUtil::DependentMap>
BuildTransitiveDependentGraph(const SchemaProto& schema) {
  SchemaUtil::DependentMap dependent_map;

  // Add all first-order dependents.
  std::unordered_set<std::string_view> known_types;
  std::unordered_set<std::string_view> unknown_types;
  for (const auto& type_config : schema.types()) {
    std::string_view schema_type(type_config.schema_type());
    if (known_types.count(schema_type) > 0) {
      return absl_ports::AlreadyExistsError(absl_ports::StrCat(
          "Field 'schema_type' '", schema_type, "' is already defined"));
    }
    known_types.insert(schema_type);
    unknown_types.erase(schema_type);
    for (const auto& property_config : type_config.properties()) {
      if (property_config.data_type() ==
          PropertyConfigProto::DataType::DOCUMENT) {
        // Need to know what schema_type these Document properties should be
        // validated against
        std::string_view property_schema_type(property_config.schema_type());
        if (known_types.count(property_schema_type) == 0) {
          unknown_types.insert(property_schema_type);
        }
        dependent_map[property_schema_type].insert(schema_type);
      }
    }
  }
  if (!unknown_types.empty()) {
    return absl_ports::InvalidArgumentError(absl_ports::StrCat(
        "Undefined 'schema_type's: ", absl_ports::StrJoin(unknown_types, ",")));
  }
  return ExpandTranstiveDependents(dependent_map);
}

libtextclassifier3::StatusOr<SchemaUtil::DependentMap> SchemaUtil::Validate(
    const SchemaProto& schema) {
  // 1. Build the dependent map. This will detect any cycles, non-existent or
  // duplicate types in the schema.
  ICING_ASSIGN_OR_RETURN(SchemaUtil::DependentMap dependent_map,
                         BuildTransitiveDependentGraph(schema));

  // Tracks PropertyConfigs within a SchemaTypeConfig that we've validated
  // already.
  std::unordered_set<std::string_view> known_property_names;

  // 2. Validate the properties of each type.
  for (const auto& type_config : schema.types()) {
    std::string_view schema_type(type_config.schema_type());
    ICING_RETURN_IF_ERROR(ValidateSchemaType(schema_type));

    // We only care about properties being unique within one type_config
    known_property_names.clear();

    for (const auto& property_config : type_config.properties()) {
      std::string_view property_name(property_config.property_name());
      ICING_RETURN_IF_ERROR(ValidatePropertyName(property_name, schema_type));

      // Property names must be unique
      if (!known_property_names.insert(property_name).second) {
        return absl_ports::AlreadyExistsError(absl_ports::StrCat(
            "Field 'property_name' '", property_name,
            "' is already defined for schema '", schema_type, "'"));
      }

      auto data_type = property_config.data_type();
      ICING_RETURN_IF_ERROR(
          ValidateDataType(data_type, schema_type, property_name));

      if (data_type == PropertyConfigProto::DataType::DOCUMENT) {
        // Need to know what schema_type these Document properties should be
        // validated against
        std::string_view property_schema_type(property_config.schema_type());
        libtextclassifier3::Status validated_status =
            ValidateSchemaType(property_schema_type);
        if (!validated_status.ok()) {
          return absl_ports::Annotate(
              validated_status,
              absl_ports::StrCat("Field 'schema_type' is required for DOCUMENT "
                                 "data_types in schema property '",
                                 schema_type, ".", property_name, "'"));
        }
      }

      ICING_RETURN_IF_ERROR(ValidateCardinality(property_config.cardinality(),
                                                schema_type, property_name));

      if (data_type == PropertyConfigProto::DataType::STRING) {
        ICING_RETURN_IF_ERROR(ValidateStringIndexingConfig(
            property_config.string_indexing_config(), data_type, schema_type,
            property_name));
      }
    }
  }

  return dependent_map;
}

libtextclassifier3::Status SchemaUtil::ValidateSchemaType(
    std::string_view schema_type) {
  // Require a schema_type
  if (schema_type.empty()) {
    return absl_ports::InvalidArgumentError(
        "Field 'schema_type' cannot be empty.");
  }

  return libtextclassifier3::Status::OK;
}

libtextclassifier3::Status SchemaUtil::ValidatePropertyName(
    std::string_view property_name, std::string_view schema_type) {
  // Require a property_name
  if (property_name.empty()) {
    return absl_ports::InvalidArgumentError(
        absl_ports::StrCat("Field 'property_name' for schema '", schema_type,
                           "' cannot be empty."));
  }

  // Only support alphanumeric values.
  for (char c : property_name) {
    if (!std::isalnum(c)) {
      return absl_ports::InvalidArgumentError(
          absl_ports::StrCat("Field 'property_name' '", property_name,
                             "' can only contain alphanumeric characters."));
    }
  }

  return libtextclassifier3::Status::OK;
}

libtextclassifier3::Status SchemaUtil::ValidateDataType(
    PropertyConfigProto::DataType::Code data_type, std::string_view schema_type,
    std::string_view property_name) {
  // UNKNOWN is the default enum value and should only be used for backwards
  // compatibility
  if (data_type == PropertyConfigProto::DataType::UNKNOWN) {
    return absl_ports::InvalidArgumentError(absl_ports::StrCat(
        "Field 'data_type' cannot be UNKNOWN for schema property '",
        schema_type, ".", property_name, "'"));
  }

  return libtextclassifier3::Status::OK;
}

libtextclassifier3::Status SchemaUtil::ValidateCardinality(
    PropertyConfigProto::Cardinality::Code cardinality,
    std::string_view schema_type, std::string_view property_name) {
  // UNKNOWN is the default enum value and should only be used for backwards
  // compatibility
  if (cardinality == PropertyConfigProto::Cardinality::UNKNOWN) {
    return absl_ports::InvalidArgumentError(absl_ports::StrCat(
        "Field 'cardinality' cannot be UNKNOWN for schema property '",
        schema_type, ".", property_name, "'"));
  }

  return libtextclassifier3::Status::OK;
}

libtextclassifier3::Status SchemaUtil::ValidateStringIndexingConfig(
    const StringIndexingConfig& config,
    PropertyConfigProto::DataType::Code data_type, std::string_view schema_type,
    std::string_view property_name) {
  if (config.term_match_type() == TermMatchType::UNKNOWN &&
      config.tokenizer_type() != StringIndexingConfig::TokenizerType::NONE) {
    // They set a tokenizer type, but no term match type.
    return absl_ports::InvalidArgumentError(absl_ports::StrCat(
        "Indexed string property '", schema_type, ".", property_name,
        "' cannot have a term match type UNKNOWN"));
  }

  if (config.term_match_type() != TermMatchType::UNKNOWN &&
      config.tokenizer_type() == StringIndexingConfig::TokenizerType::NONE) {
    // They set a term match type, but no tokenizer type
    return absl_ports::InvalidArgumentError(
        absl_ports::StrCat("Indexed string property '", property_name,
                           "' cannot have a tokenizer type of NONE"));
  }

  return libtextclassifier3::Status::OK;
}

void SchemaUtil::BuildTypeConfigMap(
    const SchemaProto& schema, SchemaUtil::TypeConfigMap* type_config_map) {
  type_config_map->clear();
  for (const SchemaTypeConfigProto& type_config : schema.types()) {
    type_config_map->emplace(type_config.schema_type(), type_config);
  }
}

SchemaUtil::ParsedPropertyConfigs SchemaUtil::ParsePropertyConfigs(
    const SchemaTypeConfigProto& type_config) {
  ParsedPropertyConfigs parsed_property_configs;

  // TODO(cassiewang): consider caching property_config_map for some properties,
  // e.g. using LRU cache. Or changing schema.proto to use go/protomap.
  for (const PropertyConfigProto& property_config : type_config.properties()) {
    parsed_property_configs.property_config_map.emplace(
        property_config.property_name(), &property_config);
    if (property_config.cardinality() ==
        PropertyConfigProto::Cardinality::REQUIRED) {
      ++parsed_property_configs.num_required_properties;
    }

    // A non-default term_match_type indicates that this property is meant to be
    // indexed.
    if (IsIndexedProperty(property_config)) {
      ++parsed_property_configs.num_indexed_properties;
    }

    // A non-default value_type indicates that this property is meant to be
    // joinable.
    if (property_config.joinable_config().value_type() !=
        JoinableConfig::ValueType::NONE) {
      ++parsed_property_configs.num_joinable_properties;
    }
  }

  return parsed_property_configs;
}

const SchemaUtil::SchemaDelta SchemaUtil::ComputeCompatibilityDelta(
    const SchemaProto& old_schema, const SchemaProto& new_schema,
    const DependentMap& new_schema_dependent_map) {
  SchemaDelta schema_delta;

  TypeConfigMap old_type_config_map, new_type_config_map;
  BuildTypeConfigMap(old_schema, &old_type_config_map);
  BuildTypeConfigMap(new_schema, &new_type_config_map);

  // Iterate through and check each field of the old schema
  for (const auto& old_type_config : old_schema.types()) {
    auto new_schema_type_and_config =
        new_type_config_map.find(old_type_config.schema_type());

    if (new_schema_type_and_config == new_type_config_map.end()) {
      // Didn't find the old schema type in the new schema, all the old
      // documents of this schema type are invalid without the schema
      ICING_VLOG(1) << absl_ports::StrCat("Previously defined schema type '",
                                          old_type_config.schema_type(),
                                          "' was not defined in new schema");
      schema_delta.schema_types_deleted.insert(old_type_config.schema_type());
      continue;
    }

    ParsedPropertyConfigs new_parsed_property_configs =
        ParsePropertyConfigs(new_schema_type_and_config->second);

    // We only need to check the old, existing properties to see if they're
    // compatible since we'll have old data that may be invalidated or need to
    // be reindexed.
    int32_t old_required_properties = 0;
    int32_t old_indexed_properties = 0;
    int32_t old_joinable_properties = 0;

    // If there is a different number of properties, then there must have been a
    // change.
    bool has_property_changed =
        old_type_config.properties_size() !=
        new_schema_type_and_config->second.properties_size();
    bool is_incompatible = false;
    bool is_index_incompatible = false;
    bool is_join_incompatible = false;
    for (const auto& old_property_config : old_type_config.properties()) {
      if (old_property_config.cardinality() ==
          PropertyConfigProto::Cardinality::REQUIRED) {
        ++old_required_properties;
      }

      // A non-default term_match_type indicates that this property is meant to
      // be indexed.
      bool is_indexed_property = IsIndexedProperty(old_property_config);
      if (is_indexed_property) {
        ++old_indexed_properties;
      }

      bool is_joinable_property =
          old_property_config.joinable_config().value_type() !=
          JoinableConfig::ValueType::NONE;
      if (is_joinable_property) {
        ++old_joinable_properties;
      }

      auto new_property_name_and_config =
          new_parsed_property_configs.property_config_map.find(
              old_property_config.property_name());

      if (new_property_name_and_config ==
          new_parsed_property_configs.property_config_map.end()) {
        // Didn't find the old property
        ICING_VLOG(1) << absl_ports::StrCat(
            "Previously defined property type '", old_type_config.schema_type(),
            ".", old_property_config.property_name(),
            "' was not defined in new schema");
        is_incompatible = true;
        is_index_incompatible |= is_indexed_property;
        is_join_incompatible |= is_joinable_property;
        continue;
      }

      const PropertyConfigProto* new_property_config =
          new_property_name_and_config->second;
      if (!has_property_changed &&
          !ArePropertiesEqual(old_property_config, *new_property_config)) {
        // Finally found a property that changed.
        has_property_changed = true;
      }

      if (!IsPropertyCompatible(old_property_config, *new_property_config)) {
        ICING_VLOG(1) << absl_ports::StrCat(
            "Property '", old_type_config.schema_type(), ".",
            old_property_config.property_name(), "' is incompatible.");
        is_incompatible = true;
      }

      // Any change in the indexed property requires a reindexing
      if (!IsTermMatchTypeCompatible(
              old_property_config.string_indexing_config(),
              new_property_config->string_indexing_config()) ||
          !IsIntegerNumericMatchTypeCompatible(
              old_property_config.integer_indexing_config(),
              new_property_config->integer_indexing_config()) ||
          old_property_config.document_indexing_config()
                  .index_nested_properties() !=
              new_property_config->document_indexing_config()
                  .index_nested_properties()) {
        is_index_incompatible = true;
      }

      if (old_property_config.joinable_config().value_type() !=
          new_property_config->joinable_config().value_type()) {
        is_join_incompatible = true;
      }
    }

    // We can't have new properties that are REQUIRED since we won't know how
    // to backfill the data, and the existing data will be invalid. We're
    // guaranteed from our previous checks that all the old properties are also
    // present in the new property config, so we can do a simple int comparison
    // here to detect new required properties.
    if (new_parsed_property_configs.num_required_properties >
        old_required_properties) {
      ICING_VLOG(1) << absl_ports::StrCat(
          "New schema '", old_type_config.schema_type(),
          "' has REQUIRED properties that are not "
          "present in the previously defined schema");
      is_incompatible = true;
    }

    // If we've gained any new indexed properties, then the section ids may
    // change. Since the section ids are stored in the index, we'll need to
    // reindex everything.
    if (new_parsed_property_configs.num_indexed_properties >
        old_indexed_properties) {
      ICING_VLOG(1) << "Set of indexed properties in schema type '"
                    << old_type_config.schema_type()
                    << "' has changed, required reindexing.";
      is_index_incompatible = true;
    }

    // If we've gained any new joinable properties, then the joinable property
    // ids may change. Since the joinable property ids are stored in the cache,
    // we'll need to reconstruct joinable cache.
    if (new_parsed_property_configs.num_joinable_properties >
        old_joinable_properties) {
      ICING_VLOG(1) << "Set of joinable properties in schema type '"
                    << old_type_config.schema_type()
                    << "' has changed, required reconstructing joinable cache.";
      is_join_incompatible = true;
    }

    if (is_incompatible) {
      AddIncompatibleChangeToDelta(schema_delta.schema_types_incompatible,
                                   old_type_config, new_schema_dependent_map,
                                   old_type_config_map, new_type_config_map);
    }

    if (is_index_incompatible) {
      AddIncompatibleChangeToDelta(schema_delta.schema_types_index_incompatible,
                                   old_type_config, new_schema_dependent_map,
                                   old_type_config_map, new_type_config_map);
    }

    if (is_join_incompatible) {
      AddIncompatibleChangeToDelta(schema_delta.schema_types_join_incompatible,
                                   old_type_config, new_schema_dependent_map,
                                   old_type_config_map, new_type_config_map);
    }

    if (!is_incompatible && !is_index_incompatible && !is_join_incompatible &&
        has_property_changed) {
      schema_delta.schema_types_changed_fully_compatible.insert(
          old_type_config.schema_type());
    }

    // Lastly, remove this type from the map. We know that this type can't
    // come up in future iterations through the old schema types because the old
    // type config has unique types.
    new_type_config_map.erase(old_type_config.schema_type());
  }

  // Any types that are still present in the new_type_config_map are newly added
  // types.
  schema_delta.schema_types_new.reserve(new_type_config_map.size());
  for (auto& kvp : new_type_config_map) {
    schema_delta.schema_types_new.insert(std::move(kvp.first));
  }

  return schema_delta;
}

}  // namespace lib
}  // namespace icing