pilosa: use different holder per db/table pair

Pilosa holder cannot be opened more than once. It has a channel to
signal opened state (currently unused) that is closed after the first
Open. This causes panics in the next calls:

    panic: close of closed channel

The driver was using one holder for all db/tables. This called
holder.Open once per table. This works well if all indexes belong to the
same db/table but called Open several times when the indexes belong to
more than one db/table pair. Now it has a map of holders so it's only
opened once.

Signed-off-by: Javi Fontan <[email protected]>
(cherry picked from commit 144b824)

Author: jfontan

sql/index/pilosa/driver.go

@@ -67,19 +67,22 @@ type (
 		timeMapping time.Duration
 	}
 
+	tableMap map[string]*pilosa.Holder
+	dbMap    map[string]tableMap
+
 	// Driver implements sql.IndexDriver interface.
 	Driver struct {
-		root   string
-		holder *pilosa.Holder
+		root    string
+		holders dbMap
 	}
 )
 
 // NewDriver returns a new instance of pilosa.Driver
 // which satisfies sql.IndexDriver interface
 func NewDriver(root string) *Driver {
 	return &Driver{
-		root:   root,
-		holder: pilosa.NewHolder(),
+		root:    root,
+		holders: make(dbMap),
 	}
 }
 
@@ -114,8 +117,9 @@ func (d *Driver) Create(
 		return nil, err
 	}
 
-	d.holder.Path = d.pilosaDirPath(db, table)
-	idx, err := d.holder.CreateIndexIfNotExists(
+	holder := d.holder(db, table)
+	holder.Path = d.pilosaDirPath(db, table)
+	idx, err := holder.CreateIndexIfNotExists(
 		indexName(db, table),
 		pilosa.IndexOptions{},
 	)
@@ -144,19 +148,20 @@ func (d *Driver) LoadAll(db, table string) ([]sql.Index, error) {
 		root    = filepath.Join(d.root, db, table)
 	)
 
-	d.holder.Path = d.pilosaDirPath(db, table)
-	if _, err := os.Stat(d.holder.Path); err != nil {
+	holder := d.holder(db, table)
+	holder.Path = d.pilosaDirPath(db, table)
+	if _, err := os.Stat(holder.Path); err != nil {
 		if os.IsNotExist(err) {
 			return indexes, nil
 		}
 		return nil, err
 	}
 
-	err := d.holder.Open()
+	err := holder.Open()
 	if err != nil {
 		return nil, err
 	}
-	defer d.holder.Close()
+	defer holder.Close()
 
 	dirs, err := ioutil.ReadDir(root)
 	if err != nil {
@@ -185,9 +190,27 @@ func (d *Driver) LoadAll(db, table string) ([]sql.Index, error) {
 	return indexes, nil
 }
 
+func (d *Driver) holder(db, table string) *pilosa.Holder {
+	tables, ok := d.holders[db]
+	if !ok {
+		tables = make(tableMap)
+		d.holders[db] = tables
+	}
+
+	holder, ok := tables[table]
+	if !ok {
+		holder = pilosa.NewHolder()
+		tables[table] = holder
+		d.holders[db] = tables
+	}
+
+	return holder
+}
+
 func (d *Driver) loadIndex(db, table, id string) (*pilosaIndex, error) {
 	name := indexName(db, table)
-	idx := d.holder.Index(name)
+	holder := d.holder(db, table)
+	idx := holder.Index(name)
 	if idx == nil {
 		return nil, errLoadingIndex.New(name)
 	}

sql/index/pilosa/driver_test.go

@@ -80,6 +80,23 @@ func TestLoadAll(t *testing.T) {
 	require.True(ok)
 
 	require.Equal(i1.index.Name(), i2.index.Name())
+
+	// Load index from another table. Previously this panicked as the same
+	// pilosa.Holder was used for all indexes.
+
+	idx3, err := d.Create("db", "table2", "id1", makeExpressions("table2", "hash1"), nil)
+	require.NoError(err)
+	it3 := &partitionKeyValueIter{
+		partitions:  2,
+		offset:      0,
+		total:       64,
+		expressions: idx3.Expressions(),
+		location:    randLocation,
+	}
+	require.NoError(d.Save(sql.NewEmptyContext(), idx3, it3))
+
+	indexes, err = d.LoadAll("db", "table2")
+	require.NoError(err)
 }
 
 type logLoc struct {