@SuppressWarnings("checkstyle:CyclomaticComplexity")
  @Override
  public void renameTable(TableIdentifier from, TableIdentifier to) {
    if (from.equals(to)) {
      return;
    }

    if (!tableExists(from)) {
      throw new NoSuchTableException("Table does not exist: %s", from);
    }

    if (!namespaceExists(to.namespace())) {
      throw new NoSuchNamespaceException("Namespace does not exist: %s", to.namespace());
    }

    if (schemaVersion == JdbcUtil.SchemaVersion.V1 && viewExists(to)) {
      throw new AlreadyExistsException("Cannot rename %s to %s. View already exists", from, to);
    }

    if (tableExists(to)) {
      throw new AlreadyExistsException("Table already exists: %s", to);
    }

    int updatedRecords =
        execute(
            err -> {
              // SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
              if (err instanceof SQLIntegrityConstraintViolationException
                  || (err.getMessage() != null && err.getMessage().contains("constraint failed"))) {
                throw new AlreadyExistsException("Table already exists: %s", to);
              }
            },
            (schemaVersion == JdbcUtil.SchemaVersion.V1)
                ? JdbcUtil.V1_RENAME_TABLE_SQL
                : JdbcUtil.V0_RENAME_TABLE_SQL,
            JdbcUtil.namespaceToString(to.namespace()),
            to.name(),
            catalogName,
            JdbcUtil.namespaceToString(from.namespace()),
            from.name());

    if (updatedRecords == 1) {
      LOG.info("Renamed table from {}, to {}", from, to);
    } else if (updatedRecords == 0) {
      throw new NoSuchTableException("Table does not exist: %s", from);
    } else {
      LOG.warn(
          "Rename operation affected {} rows: the catalog table's primary key assumption has been violated",
          updatedRecords);
    }
  }
@Test
  public void testRenameTable() {
    TableIdentifier from = TableIdentifier.of("db", "tbl1");
    TableIdentifier to = TableIdentifier.of("db", "tbl2-newtable");
    catalog.createTable(from, SCHEMA, PartitionSpec.unpartitioned());
    catalog.renameTable(from, to);
    assertThat(catalog.listTables(to.namespace())).contains(to).doesNotContain(from);
    assertThat(catalog.loadTable(to).name()).endsWith(to.name());

    assertThatThrownBy(() -> catalog.renameTable(TableIdentifier.of("db", "tbl-not-exists"), to))
        .isInstanceOf(NoSuchTableException.class)
        .hasMessage("Table does not exist: db.tbl-not-exists");

    // rename table to existing table name!
    TableIdentifier from2 = TableIdentifier.of("db", "tbl2");
    catalog.createTable(from2, SCHEMA, PartitionSpec.unpartitioned());
    assertThatThrownBy(() -> catalog.renameTable(from2, to))
        .isInstanceOf(AlreadyExistsException.class)
        .hasMessage("Table already exists: db.tbl2-newtable");
  }