sql/plan: make sure outdated indexes can be dropped

sql/index.go

@@ -253,11 +253,12 @@ func (r *IndexRegistry) LoadIndexes(dbs Databases) error {
 						r.statuses[k] = IndexReady
 					} else {
 						logrus.Warnf(
-							"index %q is outdated and will not be used, you can remove it using `DROP INDEX %s`",
+							"index %q is outdated and will not be used, you can remove it using `DROP INDEX %s ON %s`",
 							idx.ID(),
 							idx.ID(),
+							idx.Table(),
 						)
-						r.statuses[k] = IndexOutdated
+						r.MarkOutdated(idx)
 					}
 				}
 			}
@@ -267,6 +268,12 @@ func (r *IndexRegistry) LoadIndexes(dbs Databases) error {
 	return nil
 }
 
+// MarkOutdated sets the index status as outdated. This method is not thread
+// safe and should not be used directly except for testing.
+func (r *IndexRegistry) MarkOutdated(idx Index) {
+	r.statuses[indexKey{idx.Database(), idx.ID()}] = IndexOutdated
+}
+
 func (r *IndexRegistry) retainIndex(db, id string) {
 	r.rcmut.Lock()
 	defer r.rcmut.Unlock()
@@ -336,7 +343,7 @@ func (r *IndexRegistry) IndexesByTable(db, table string) []Index {
 	r.mut.RLock()
 	defer r.mut.RUnlock()
 
-	indexes := []Index{}
+	var indexes []Index
 	for _, key := range r.indexOrder {
 		idx := r.indexes[key]
 		if idx.Database() == db && idx.Table() == table {

sql/plan/drop_index.go

@@ -62,7 +62,7 @@ func (d *DropIndex) RowIter(ctx *sql.Context) (sql.RowIter, error) {
 	}
 	d.Catalog.ReleaseIndex(index)
 
-	if !d.Catalog.CanUseIndex(index) {
+	if !d.Catalog.CanRemoveIndex(index) {
 		return nil, ErrIndexNotAvailable.New(d.Name)
 	}
 

sql/plan/drop_index_test.go

@@ -97,3 +97,47 @@ func TestDeleteIndexNotReady(t *testing.T) {
 	close(done)
 	<-ready
 }
+
+func TestDeleteIndexOutdated(t *testing.T) {
+	require := require.New(t)
+
+	table := mem.NewTable("foo", sql.Schema{
+		{Name: "a", Source: "foo"},
+		{Name: "b", Source: "foo"},
+		{Name: "c", Source: "foo"},
+	})
+
+	driver := new(mockDriver)
+	catalog := sql.NewCatalog()
+	catalog.RegisterIndexDriver(driver)
+	db := mem.NewDatabase("foo")
+	db.AddTable("foo", table)
+	catalog.AddDatabase(db)
+
+	var expressions = []sql.Expression{
+		expression.NewGetFieldWithTable(0, sql.Int64, "foo", "c", true),
+		expression.NewGetFieldWithTable(1, sql.Int64, "foo", "a", true),
+	}
+
+	done, ready, err := catalog.AddIndex(&mockIndex{id: "idx", db: "foo", table: "foo", exprs: expressions})
+	require.NoError(err)
+	close(done)
+	<-ready
+
+	idx := catalog.Index("foo", "idx")
+	require.NotNil(idx)
+	catalog.ReleaseIndex(idx)
+	catalog.MarkOutdated(idx)
+
+	di := NewDropIndex("idx", NewResolvedTable(table))
+	di.Catalog = catalog
+	di.CurrentDatabase = "foo"
+
+	_, err = di.RowIter(sql.NewEmptyContext())
+	require.NoError(err)
+
+	time.Sleep(50 * time.Millisecond)
+
+	require.Equal([]string{"idx"}, driver.deleted)
+	require.Nil(catalog.Index("foo", "idx"))
+}