func()

in cassandra-bigtable-migration-tools/cassandra-bigtable-proxy/bigtable/bigtable.go [250:307]

• 48 lines of code
• 11 McCabe index (conditional complexity)

func (btc *BigtableClient) DropTable(ctx context.Context, data *translator.DropTableStatementMap, schemaMappingTableName string) error {
	client, ok := btc.Clients[data.Keyspace]
	if !ok {
		return fmt.Errorf("invalid keyspace `%s`", data.Keyspace)
	}
	adminClient, ok := btc.AdminClients[data.Keyspace]
	if !ok {
		return fmt.Errorf("invalid keyspace `%s`", data.Keyspace)
	}

	// first clean up table from schema mapping table because that's the SoT
	tbl := client.Open(schemaMappingTableName)
	var deleteMuts []*bigtable.Mutation
	var rowKeysToDelete []string
	err := tbl.ReadRows(ctx, bigtable.PrefixRange(data.Table+"#"), func(row bigtable.Row) bool {
		mut := bigtable.NewMutation()
		mut.DeleteRow()
		deleteMuts = append(deleteMuts, mut)
		rowKeysToDelete = append(rowKeysToDelete, row.Key())
		return true
	})

	if err != nil {
		return err
	}

	btc.Logger.Info("drop table: deleting schema rows")
	_, err = tbl.ApplyBulk(ctx, rowKeysToDelete, deleteMuts)
	if err != nil {
		return err
	}

	// do a read to check if the table exists to save on admin API write quota
	exists, err := btc.tableResourceExists(ctx, adminClient, data.Table)
	if err != nil {
		return err
	}
	if exists {
		btc.Logger.Info("drop table: deleting bigtable table")
		err = adminClient.DeleteTable(ctx, data.Table)
	}
	if err != nil {
		return err
	}

	// this error behavior is done independently of the table resource existing or not because the schema mapping table is the SoT, not the table resource
	if len(rowKeysToDelete) == 0 && !data.IfExists {
		return fmt.Errorf("cannot delete table %s because it does not exist", data.Table)
	}

	btc.Logger.Info("reloading schema mappings")
	err = btc.reloadSchemaMappings(ctx, data.Keyspace, schemaMappingTableName)
	if err != nil {
		return err
	}

	return nil
}