void InformationSchemaCatalog::FillIndexColumnsTable()

in backend/query/information_schema_catalog.cc [1280:1416]


void InformationSchemaCatalog::FillIndexColumnsTable() {
  auto index_columns =
      tables_by_name_.at(GetNameForDialect(kIndexColumns)).get();

  // Add table rows.
  std::vector<std::vector<zetasql::Value>> rows;
  for (const Table* table : default_schema_->tables()) {
    const auto& [table_schema_part, table_name_part] =
        GetSchemaAndNameForInformationSchema(table->Name());
    // Add normal indexes.
    for (const Index* index : table->indexes()) {
      int pos = 1;
      // Add key columns.
      for (const KeyColumn* key_column : index->key_columns()) {
        rows.push_back({// table_catalog
                        DialectTableCatalog(),
                        // table_schema
                        String(table_schema_part),
                        // table_name
                        String(table_name_part),
                        // index_name
                        String(SDLObjectName::GetInSchemaName(index->Name())),
                        // index_type
                        String(kIndex),
                        // column_name
                        String(key_column->column()->Name()),
                        // ordinal_position
                        Int64(pos++),
                        // column_ordering
                        DialectColumnOrdering(key_column),
                        // is_nullable
                        String(key_column->column()->is_nullable() &&
                                       !index->is_null_filtered()
                                   ? kYes
                                   : kNo),
                        // spanner_type
                        GetSpannerType(key_column->column())});
      }
      // Add storing columns.
      for (const Column* column : index->stored_columns()) {
        rows.push_back({
            // table_catalog
            DialectTableCatalog(),
            // table_schema
            String(table_schema_part),
            // table_name
            String(table_name_part),
            // index_name
            String(SDLObjectName::GetInSchemaName(index->Name())),
            // index_type
            String(kIndex),
            // column_name
            String(column->Name()),
            // ordinal_position
            NullInt64(),
            // column_ordering
            NullString(),
            // is_nullable
            String(column->is_nullable() ? kYes : kNo),
            // spanner_type
            GetSpannerType(column),
        });
      }
    }

    // Add the primary key columns.
    {
      int pos = 1;
      for (const KeyColumn* key_column : table->primary_key()) {
        rows.push_back({
            // table_catalog
            DialectTableCatalog(),
            // table_schema
            String(table_schema_part),
            // table_name
            String(table_name_part),
            // index_name
            String(kPrimary_Key),
            // index_type
            String(kPrimary_Key),
            // column_name
            String(key_column->column()->Name()),
            // ordinal_position
            Int64(pos++),
            // column_ordering
            DialectColumnOrdering(key_column),
            // is_nullable
            String(key_column->column()->is_nullable() ? kYes : kNo),
            // spanner_type
            GetSpannerType(key_column->column()),
        });
      }
    }
  }

  // The primary key columns for tables in the INFORMATION_SCHEMA are not added
  // in the PG dialect in production so we also don't add it in the emulator.
  if (dialect_ != DatabaseDialect::POSTGRESQL) {
    // Add the information schema primary key columns.
    for (const auto& table : this->tables()) {
      int primary_key_ordinal = 1;
      for (int i = 0; i < table->NumColumns(); ++i) {
        const auto* column = table->GetColumn(i);
        const auto* metadata = FindKeyColumnMetadata(dialect_, table, column);
        if (metadata == nullptr) {
          continue;  // Not a primary key column.
        }
        rows.push_back({
            // table_catalog
            DialectTableCatalog(),
            // table_schema
            String(kInformationSchema),
            // table_name
            String(table->Name()),
            // index_name
            String(kPrimary_Key),
            // index_type
            String(kPrimary_Key),
            // column_name
            String(column->Name()),
            // ordinal_position
            Int64(metadata->primary_key_ordinal > 0
                      ? metadata->primary_key_ordinal
                      : primary_key_ordinal++),
            // column_ordering
            String(metadata->column_ordering),
            // is_nullable
            String(metadata->is_nullable),
            // spanner_type
            String(metadata->spanner_type),
        });
      }
    }
  }

  index_columns->SetContents(rows);
}