private void listForeignKeys()

in v1/src/main/java/com/google/cloud/teleport/spanner/ddl/InformationSchemaScanner.java [778:888]
109 lines of code
9 McCabe index (conditional complexity)

  private void listForeignKeys(Map<String, NavigableMap<String, ForeignKey.Builder>> foreignKeys) {
    Statement statement;

    switch (dialect) {
      case GOOGLE_STANDARD_SQL:
        statement =
            Statement.of(
                "SELECT rc.constraint_name,"
                    + " kcu1.table_schema,"
                    + " kcu1.table_name,"
                    + " kcu1.column_name,"
                    + " kcu2.table_schema,"
                    + " kcu2.table_name,"
                    + " kcu2.column_name,"
                    + " rc.delete_rule,"
                    + " tc.enforced"
                    + " FROM information_schema.referential_constraints as rc"
                    + " INNER JOIN information_schema.table_constraints as tc"
                    + " ON tc.constraint_catalog = rc.constraint_catalog"
                    + " AND tc.constraint_schema = rc.constraint_schema"
                    + " AND tc.constraint_name = rc.constraint_name"
                    + " INNER JOIN information_schema.key_column_usage as kcu1"
                    + " ON kcu1.constraint_catalog = rc.constraint_catalog"
                    + " AND kcu1.constraint_schema = rc.constraint_schema"
                    + " AND kcu1.constraint_name = rc.constraint_name"
                    + " INNER JOIN information_schema.key_column_usage as kcu2"
                    + " ON kcu2.constraint_catalog = rc.unique_constraint_catalog"
                    + " AND kcu2.constraint_schema = rc.unique_constraint_schema"
                    + " AND kcu2.constraint_name = rc.unique_constraint_name"
                    + " AND kcu2.ordinal_position = kcu1.position_in_unique_constraint"
                    + " WHERE rc.constraint_catalog = kcu1.constraint_catalog"
                    + " AND rc.constraint_catalog = kcu2.constraint_catalog"
                    + " AND rc.constraint_schema NOT IN "
                    + " ('INFORMATION_SCHEMA', 'SPANNER_SYS')"
                    + " ORDER BY rc.constraint_name, kcu1.ordinal_position;");
        break;
      case POSTGRESQL:
        statement =
            Statement.of(
                "SELECT rc.constraint_name,"
                    + " kcu1.table_schema,"
                    + " kcu1.table_name,"
                    + " kcu1.column_name,"
                    + " kcu2.table_schema,"
                    + " kcu2.table_name,"
                    + " kcu2.column_name,"
                    + " rc.delete_rule,"
                    + " tc.enforced"
                    + " FROM information_schema.referential_constraints as rc"
                    + " INNER JOIN information_schema.table_constraints as tc"
                    + " ON tc.constraint_catalog = rc.constraint_catalog"
                    + " AND tc.constraint_schema = rc.constraint_schema"
                    + " AND tc.constraint_name = rc.constraint_name"
                    + " INNER JOIN information_schema.key_column_usage as kcu1"
                    + " ON kcu1.constraint_catalog = rc.constraint_catalog"
                    + " AND kcu1.constraint_schema = rc.constraint_schema"
                    + " AND kcu1.constraint_name = rc.constraint_name"
                    + " INNER JOIN information_schema.key_column_usage as kcu2"
                    + " ON kcu2.constraint_catalog = rc.unique_constraint_catalog"
                    + " AND kcu2.constraint_schema = rc.unique_constraint_schema"
                    + " AND kcu2.constraint_name = rc.unique_constraint_name"
                    + " AND kcu2.ordinal_position = kcu1.position_in_unique_constraint"
                    + " WHERE rc.constraint_catalog = kcu1.constraint_catalog"
                    + " AND rc.constraint_catalog = kcu2.constraint_catalog"
                    + " AND rc.constraint_schema NOT IN "
                    + " ('information_schema', 'spanner_sys', 'pg_catalog')"
                    + " ORDER BY rc.constraint_name, kcu1.ordinal_position;");
        break;
      default:
        throw new IllegalArgumentException("Unrecognized dialect: " + dialect);
    }

    ResultSet resultSet = context.executeQuery(statement);
    while (resultSet.next()) {
      String name = resultSet.getString(0);
      String table = getQualifiedName(resultSet.getString(1), resultSet.getString(2));
      String column = resultSet.getString(3);
      String referencedTable = getQualifiedName(resultSet.getString(4), resultSet.getString(5));
      String referencedColumn = resultSet.getString(6);
      String deleteRule = resultSet.getString(7);
      String enforced = dialect == Dialect.GOOGLE_STANDARD_SQL ? resultSet.getString(8) : null;
      Map<String, ForeignKey.Builder> tableForeignKeys =
          foreignKeys.computeIfAbsent(table, k -> Maps.newTreeMap());
      ForeignKey.Builder foreignKey =
          tableForeignKeys.computeIfAbsent(
              name,
              k ->
                  ForeignKey.builder(dialect)
                      .name(name)
                      .table(table)
                      .referencedTable(referencedTable));
      if (!isNullOrEmpty(deleteRule)) {
        foreignKey.referentialAction(
            Optional.of(ReferentialAction.getReferentialAction("DELETE", deleteRule)));
      }
      if (!isNullOrEmpty(enforced)) {
        switch (enforced.trim().toUpperCase()) {
          case "YES":
            foreignKey.isEnforced(true);
            break;
          case "NO":
            foreignKey.isEnforced(false);
            break;
          default:
            throw new IllegalArgumentException("Illegal enforcement: " + enforced);
        }
      }
      foreignKey.columnsBuilder().add(column);
      foreignKey.referencedColumnsBuilder().add(referencedColumn);
    }
  }