tensorflow/core/tpu/graph_rewrite/combine_tpu_embedding_load_retrieve_pass.cc
// Gets load or retrieve parameters nodes, plus some other related information,
// from those candidate nodes found on a given device. There is required to be
// exactly one such node for each table in the embedding layer configuration.
//
// Parameters:
//   candidate_nodes: Load or retrieve nodes; all must have the correct device
//     and shard_id values.
//   tpu_embedding_config: Embedding layer configuration to use.
//   node_name_format: Format (printf-style) for expected node names, with %s
//     replaced by optimization algorithm name; must come from a trusted source.
//   nodes_per_table: Used to return a node for each table mentioned in
//     tpu_embedding_config; (*nodes_per_table)[i] will contain the candidate
//     node with table_id == i or table_name == table i's name.
//   is_debug_load_retrieve_node: Used to return a bool for each table that
//     tells whether a GradAccumDebug op was used for it; those ops allow the
//     user to explicitly save and restore otherwise-hidden gradient
//     accumulation buffers.
//   num_shards: Used to return number of shards from first matching node
//     (checked for consistency with others).
//
// Returns: Status (OK if successful; otherwise, an error status).
//
Gets load or retrieve parameters nodes, plus some other related information,
from those candidate nodes found on a given device. There is required to be
exactly one such node for each table in the embedding layer configuration.

Load or retrieve nodes; all must have the correct device
and shard_id values.
tpu_embedding_config: Embedding layer configuration to use.
node_name_format: Format (printf-style) for expected node names, with %s
replaced by optimization algorithm name; must come from a trusted source.
nodes_per_table: Used to return a node for each table mentioned in
tpu_embedding_config; (*nodes_per_table)[i] will contain the candidate
node with table_id == i or table_name == table i's name.
is_debug_load_retrieve_node: Used to return a bool for each table that
tells whether a GradAccumDebug op was used for it; those ops allow the
user to explicitly save and restore otherwise-hidden gradient
accumulation buffers.
num_shards: Used to return number of shards from first matching node
(checked for consistency with others).

Status (OK if successful; otherwise, an error status).
Gets load or retrieve parameters nodes plus some other related information from those candidate nodes found on a given device . There is required to be exactly one such node for each table in the embedding layer configuration . Load or retrieve nodes ; all must have the correct device and shard_id values . tpu_embedding_config : Embedding layer configuration to use . node_name_format : Format ( printf - style ) for expected node names with %s replaced by optimization algorithm name ; must come from a trusted source . nodes_per_table : Used to return a node for each table mentioned in tpu_embedding_config ; ( * nodes_per_table ) [ i ] will contain the candidate node with table_id == i or table_name == table i ' s name . is_debug_load_retrieve_node : Used to return a bool for each table that tells whether a GradAccumDebug op was used for it ; those ops allow the user to explicitly save and restore otherwise - hidden gradient accumulation buffers . num_shards : Used to return number of shards from first matching node ( checked for consistency with others ) . Status ( OK if successful ; otherwise an error status ) .
Status GetLoadOrRetrieveNodesByTable(
    const absl::flat_hash_set<Node*>& candidate_nodes,
    const tensorflow::tpu::TPUEmbeddingConfiguration& tpu_embedding_config,
    const TableNameToIntegerMap& table_name_to_id_map,
    const std::string& operation_str, std::vector<Node*>* nodes_per_table,
    std::vector<bool>* is_debug_load_retrieve_node, int* num_shards) {
  TF_RET_CHECK(!candidate_nodes.empty());
  TF_RETURN_IF_ERROR(
      GetNodeAttr((*candidate_nodes.begin())->def(), "num_shards", num_shards));
  for (Node* n : candidate_nodes) {
    int test_num_shards;
    TF_RETURN_IF_ERROR(GetNodeAttr(n->def(), "num_shards", &test_num_shards));
    TF_RET_CHECK(*num_shards == test_num_shards);
  }
  const int num_tables = tpu_embedding_config.table_descriptor_size();
  nodes_per_table->clear();
  nodes_per_table->resize(num_tables, nullptr);
  is_debug_load_retrieve_node->clear();
  is_debug_load_retrieve_node->resize(num_tables, false);
  for (Node* n : candidate_nodes) {
    int table_id;
    TF_RETURN_IF_ERROR(GetNodeAttr(n->def(), "table_id", &table_id));
    std::string table_name;
    TF_RETURN_IF_ERROR(GetNodeAttr(n->def(), "table_name", &table_name));
    if (table_id < 0 && table_name.empty()) {
      return errors::InvalidArgument(
          "Neither table_id nor table_name attribute specified in node " +
          n->name());
    }
    if (table_id >= 0 && !table_name.empty()) {
      return errors::InvalidArgument(
          "Both table_id and table_name attributes specified in node " +
          n->name());
    }
    if (!table_name.empty()) {
      if (!table_name_to_id_map.contains(table_name)) {
        return errors::InvalidArgument(
            "Table name attribute refers to non-existent table '" + table_name +
            "' in node " + n->name());
      }
      table_id = table_name_to_id_map.at(table_name);
    }
    if (table_id < 0 || table_id >= num_tables) {
      return errors::InvalidArgument(
          "table_id attribute out of range in node " + n->name());
    }
    if ((*nodes_per_table)[table_id] != nullptr) {
      return errors::AlreadyExists("Found duplicate table_id caused by op " +
                                   n->name());
    }
    const auto alg = tpu_embedding_config.table_descriptor(table_id)
                         .optimization_parameters()
                         .parameters_case();
    const std::string alg_name = tpu::GetOptimizationAlgorithmName(alg);
    const std::string expected_op_name =
        absl::StrCat(operation_str, "TPUEmbedding", alg_name, "Parameters");
    const std::string expected_op_name_debug =
        absl::StrCat(expected_op_name, "GradAccumDebug");
    if (n->op_def().name() != expected_op_name &&
        n->op_def().name() != expected_op_name_debug) {
      return errors::InvalidArgument(
          absl::StrFormat("Node %s has op type %s instead of the name %s or %s "
                          "expected from the embedding layer configuration",
                          n->name(), n->op_def().name(), expected_op_name,
                          expected_op_name_debug));
    }
    (*nodes_per_table)[table_id] = n;
    (*is_debug_load_retrieve_node)[table_id] =
        (n->op_def().name() == expected_op_name_debug);
  }
  for (int table_id = 0; table_id < num_tables; ++table_id) {
    if ((*nodes_per_table)[table_id] == nullptr) {
      return errors::NotFound(absl::StrFormat(
          "Did not find per-table load or retrieve op for table '%s' ID(%d)",
          tpu_embedding_config.table_descriptor(table_id).name(), table_id));
    }
  }
  return Status::OK();
}
Status GetLoadOrRetrieveNodesByTable ( const absl :: flat_hash_set < Node * > & candidate_nodes , const tensorflow :: tpu :: TPUEmbeddingConfiguration & tpu_embedding_config , const TableNameToIntegerMap & table_name_to_id_map , const std :: string & operation_str , std :: vector < Node * > * nodes_per_table , std :: vector < bool > * is_debug_load_retrieve_node , int * num_shards ) { TF_RET_CHECK ( ! candidate_nodes . empty ( ) ) ; TF_RETURN_IF_ERROR ( GetNodeAttr ( ( * candidate_nodes . begin ( ) ) -> def ( ) , " " , num_shards ) ) ; for ( Node * n : candidate_nodes ) { int test_num_shards ; TF_RETURN_IF_ERROR ( GetNodeAttr ( n -> def ( ) , " " , & test_num_shards ) ) ; TF_RET_CHECK ( * num_shards == test_num_shards ) ; } const int num_tables = tpu_embedding_config . table_descriptor_size ( ) ; nodes_per_table -> clear ( ) ; nodes_per_table -> resize ( num_tables , nullptr ) ; is_debug_load_retrieve_node -> clear ( ) ; is_debug_load_retrieve_node -> resize ( num_tables , false ) ; for ( Node * n : candidate_nodes ) { int table_id ; TF_RETURN_IF_ERROR ( GetNodeAttr ( n -> def ( ) , " " , & table_id ) ) ; std :: string table_name ; TF_RETURN_IF_ERROR ( GetNodeAttr ( n -> def ( ) , " " , & table_name ) ) ; if ( table_id < 0 && table_name . empty ( ) ) { return errors :: InvalidArgument ( " " + n -> name ( ) ) ; } if ( table_id >= 0 && ! table_name . empty ( ) ) { return errors :: InvalidArgument ( " " + n -> name ( ) ) ; } if ( ! table_name . empty ( ) ) { if ( ! table_name_to_id_map . contains ( table_name ) ) { return errors :: InvalidArgument ( " " + table_name + " " + n -> name ( ) ) ; } table_id = table_name_to_id_map . at ( table_name ) ; } if ( table_id < 0 || table_id >= num_tables ) { return errors :: InvalidArgument ( " " + n -> name ( ) ) ; } if ( ( * nodes_per_table ) [ table_id ] != nullptr ) { return errors :: AlreadyExists ( " " + n -> name ( ) ) ; } const auto alg = tpu_embedding_config . table_descriptor ( table_id ) . optimization_parameters ( ) . parameters_case ( ) ; const std :: string alg_name = tpu :: GetOptimizationAlgorithmName ( alg ) ; const std :: string expected_op_name = absl :: StrCat ( operation_str , " " , alg_name , " " ) ; const std :: string expected_op_name_debug = absl :: StrCat ( expected_op_name , " " ) ; if ( n -> op_def ( ) . name ( ) != expected_op_name && n -> op_def ( ) . name ( ) != expected_op_name_debug ) { return errors :: InvalidArgument ( absl :: StrFormat ( " " " " , n -> name ( ) , n -> op_def ( ) . name ( ) , expected_op_name , expected_op_name_debug ) ) ; } ( * nodes_per_table ) [ table_id ] = n ; ( * is_debug_load_retrieve_node ) [ table_id ] = ( n -> op_def ( ) . name ( ) == expected_op_name_debug ) ; } for ( int table_id = 0 ; table_id < num_tables ; ++ table_id ) { if ( ( * nodes_per_table ) [ table_id ] == nullptr ) { return errors :: NotFound ( absl :: StrFormat ( " " , tpu_embedding_config . table_descriptor ( table_id ) . name ( ) , table_id ) ) ; } } return Status :: OK ( ) ; }
Gets load or retrieve parameters nodes, plus some other related information,
from those candidate nodes found on a given device.
Gets load or retrieve parameters nodes plus some other related information from those candidate nodes found on a given device .
// Check that all tables have the same shard count.