diff --git a/docs/architecture.md b/docs/architecture.md
index e1114b0e42..07e799ee19 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -145,4 +145,4 @@ The interface works somewhat differently across the supported databases:
 
 A set of schemas are used to map the matchers and label sets used on reads and writes to the chunk store into appropriate operations on the index. Schemas have been added as Cortex has evolved, mainly in an attempt to better load balance writes and improve query performance.
 
-> The current schema recommendation is the **v10 schema**.
+> The current schema recommendation is the **v10 schema**. v11 schema is an experimental schema.
diff --git a/docs/single-process-config.yaml b/docs/single-process-config.yaml
index 2d1847fcb0..ec4e464a63 100644
--- a/docs/single-process-config.yaml
+++ b/docs/single-process-config.yaml
@@ -51,7 +51,7 @@ ingester:
 # for the chunks.
 schema:
   configs:
-  - from: 2019-03-25
+  - from: 2019-07-29
     store: boltdb
     object_store: filesystem
     schema: v10
@@ -65,4 +65,3 @@ storage:
 
   filesystem:
     directory: /tmp/cortex/chunks
-
diff --git a/pkg/chunk/chunk_store_test.go b/pkg/chunk/chunk_store_test.go
index 00e943bff2..6951f80881 100644
--- a/pkg/chunk/chunk_store_test.go
+++ b/pkg/chunk/chunk_store_test.go
@@ -37,6 +37,7 @@ var schemas = []struct {
 	{"v6", true},
 	{"v9", true},
 	{"v10", true},
+	{"v11", true},
 }
 
 var stores = []struct {
@@ -416,11 +417,11 @@ func TestChunkStore_LabelNamesForMetricName(t *testing.T) {
 	}{
 		{
 			`foo`,
-			[]string{"bar", "flip", "toms"},
+			[]string{labels.MetricName, "bar", "flip", "toms"},
 		},
 		{
 			`bar`,
-			[]string{"bar", "toms"},
+			[]string{labels.MetricName, "bar", "toms"},
 		},
 	} {
 		for _, schema := range schemas {
diff --git a/pkg/chunk/chunk_store_utils.go b/pkg/chunk/chunk_store_utils.go
index 9a5ef76ea3..1d83a0600b 100644
--- a/pkg/chunk/chunk_store_utils.go
+++ b/pkg/chunk/chunk_store_utils.go
@@ -42,11 +42,9 @@ func labelNamesFromChunks(chunks []Chunk) []string {
 	var result []string
 	for _, c := range chunks {
 		for _, l := range c.Metric {
-			if l.Name != model.MetricNameLabel {
-				if _, ok := keys[string(l.Name)]; !ok {
-					keys[string(l.Name)] = struct{}{}
-					result = append(result, string(l.Name))
-				}
+			if _, ok := keys[string(l.Name)]; !ok {
+				keys[string(l.Name)] = struct{}{}
+				result = append(result, string(l.Name))
 			}
 		}
 	}
diff --git a/pkg/chunk/composite_store.go b/pkg/chunk/composite_store.go
index f22aad0705..3680dc67ef 100644
--- a/pkg/chunk/composite_store.go
+++ b/pkg/chunk/composite_store.go
@@ -55,7 +55,7 @@ func (c *CompositeStore) AddPeriod(storeCfg StoreConfig, cfg PeriodConfig, index
 	var store Store
 	var err error
 	switch cfg.Schema {
-	case "v9", "v10":
+	case "v9", "v10", "v11":
 		store, err = newSeriesStore(storeCfg, schema, index, chunks, limits)
 	default:
 		store, err = newStore(storeCfg, schema, index, chunks, limits)
diff --git a/pkg/chunk/inmemory_storage_client.go b/pkg/chunk/inmemory_storage_client.go
index 56251a19f3..80464e3a7c 100644
--- a/pkg/chunk/inmemory_storage_client.go
+++ b/pkg/chunk/inmemory_storage_client.go
@@ -165,6 +165,7 @@ func (m *MockStorage) BatchWrite(ctx context.Context, batch WriteBatch) error {
 			itemComponents := decodeRangeKey(items[i].rangeValue)
 			if !bytes.Equal(itemComponents[3], metricNameRangeKeyV1) &&
 				!bytes.Equal(itemComponents[3], seriesRangeKeyV1) &&
+				!bytes.Equal(itemComponents[3], labelNamesRangeKeyV1) &&
 				!bytes.Equal(itemComponents[3], labelSeriesRangeKeyV1) {
 				return fmt.Errorf("Dupe write")
 			}
diff --git a/pkg/chunk/schema.go b/pkg/chunk/schema.go
index 8e2f5b8465..778bbb3e4c 100644
--- a/pkg/chunk/schema.go
+++ b/pkg/chunk/schema.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 )
@@ -22,6 +23,8 @@ var (
 	// For v9 schema
 	seriesRangeKeyV1      = []byte{'7'}
 	labelSeriesRangeKeyV1 = []byte{'8'}
+	// For v11 schema
+	labelNamesRangeKeyV1 = []byte{'9'}
 
 	// ErrNotSupported when a schema doesn't support that particular lookup.
 	ErrNotSupported = errors.New("not supported")
@@ -45,6 +48,8 @@ type Schema interface {
 
 	// If the query resulted in series IDs, use this method to find chunks.
 	GetChunksForSeries(from, through model.Time, userID string, seriesID []byte) ([]IndexQuery, error)
+	// Returns queries to retrieve all label names of multiple series by id.
+	GetLabelNamesForSeries(from, through model.Time, userID string, seriesID []byte) ([]IndexQuery, error)
 }
 
 // IndexQuery describes a query for entries
@@ -196,6 +201,20 @@ func (s schema) GetChunksForSeries(from, through model.Time, userID string, seri
 	return result, nil
 }
 
+func (s schema) GetLabelNamesForSeries(from, through model.Time, userID string, seriesID []byte) ([]IndexQuery, error) {
+	var result []IndexQuery
+
+	buckets := s.buckets(from, through, userID)
+	for _, bucket := range buckets {
+		entries, err := s.entries.GetLabelNamesForSeries(bucket, seriesID)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, entries...)
+	}
+	return result, nil
+}
+
 type entries interface {
 	GetWriteEntries(bucket Bucket, metricName string, labels labels.Labels, chunkID string) ([]IndexEntry, error)
 	GetLabelWriteEntries(bucket Bucket, metricName string, labels labels.Labels, chunkID string) ([]IndexEntry, error)
@@ -205,6 +224,7 @@ type entries interface {
 	GetReadMetricLabelQueries(bucket Bucket, metricName string, labelName string) ([]IndexQuery, error)
 	GetReadMetricLabelValueQueries(bucket Bucket, metricName string, labelName string, labelValue string) ([]IndexQuery, error)
 	GetChunksForSeries(bucket Bucket, seriesID []byte) ([]IndexQuery, error)
+	GetLabelNamesForSeries(bucket Bucket, seriesID []byte) ([]IndexQuery, error)
 }
 
 // original entries:
@@ -276,6 +296,10 @@ func (originalEntries) GetChunksForSeries(_ Bucket, _ []byte) ([]IndexQuery, err
 	return nil, ErrNotSupported
 }
 
+func (originalEntries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
 // v3Schema went to base64 encoded label values & a version ID
 // - range key: <label name>\0<base64(label value)>\0<chunk name>\0<version 1>
 
@@ -391,6 +415,10 @@ func (labelNameInHashKeyEntries) GetChunksForSeries(_ Bucket, _ []byte) ([]Index
 	return nil, ErrNotSupported
 }
 
+func (labelNameInHashKeyEntries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
 // v5 schema is an extension of v4, with the chunk end time in the
 // range key to improve query latency.  However, it did it wrong
 // so the chunk end times are ignored.
@@ -461,6 +489,10 @@ func (v5Entries) GetChunksForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
 	return nil, ErrNotSupported
 }
 
+func (v5Entries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
 // v6Entries fixes issues with v5 time encoding being wrong (see #337), and
 // moves label value out of range key (see #199).
 type v6Entries struct{}
@@ -537,6 +569,10 @@ func (v6Entries) GetChunksForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
 	return nil, ErrNotSupported
 }
 
+func (v6Entries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
 // v9Entries adds a layer of indirection between labels -> series -> chunks.
 type v9Entries struct {
 }
@@ -632,6 +668,10 @@ func (v9Entries) GetChunksForSeries(bucket Bucket, seriesID []byte) ([]IndexQuer
 	}, nil
 }
 
+func (v9Entries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
 // v10Entries builds on v9 by sharding index rows to reduce their size.
 type v10Entries struct {
 	rowShards uint32
@@ -736,3 +776,72 @@ func (v10Entries) GetChunksForSeries(bucket Bucket, seriesID []byte) ([]IndexQue
 		},
 	}, nil
 }
+
+func (v10Entries) GetLabelNamesForSeries(_ Bucket, _ []byte) ([]IndexQuery, error) {
+	return nil, ErrNotSupported
+}
+
+// v11Entries builds on v10 but adds index entries for each series to store respective labels.
+type v11Entries struct {
+	v10Entries
+}
+
+func (s v11Entries) GetLabelWriteEntries(bucket Bucket, metricName string, labels labels.Labels, chunkID string) ([]IndexEntry, error) {
+	seriesID := labelsSeriesID(labels)
+
+	// read first 32 bits of the hash and use this to calculate the shard
+	shard := binary.BigEndian.Uint32(seriesID) % s.rowShards
+
+	labelNames := make([]string, 0, len(labels))
+	for _, l := range labels {
+		if l.Name == model.MetricNameLabel {
+			continue
+		}
+		labelNames = append(labelNames, l.Name)
+	}
+	data, err := jsoniter.ConfigFastest.Marshal(labelNames)
+	if err != nil {
+		return nil, err
+	}
+	entries := []IndexEntry{
+		// Entry for metricName -> seriesID
+		{
+			TableName:  bucket.tableName,
+			HashValue:  fmt.Sprintf("%02d:%s:%s", shard, bucket.hashKey, metricName),
+			RangeValue: encodeRangeKey(seriesID, nil, nil, seriesRangeKeyV1),
+		},
+		// Entry for seriesID -> label names
+		{
+			TableName:  bucket.tableName,
+			HashValue:  string(seriesID),
+			RangeValue: encodeRangeKey(nil, nil, nil, labelNamesRangeKeyV1),
+			Value:      data,
+		},
+	}
+
+	// Entries for metricName:labelName -> hash(value):seriesID
+	// We use a hash of the value to limit its length.
+	for _, v := range labels {
+		if v.Name == model.MetricNameLabel {
+			continue
+		}
+		valueHash := sha256bytes(v.Value)
+		entries = append(entries, IndexEntry{
+			TableName:  bucket.tableName,
+			HashValue:  fmt.Sprintf("%02d:%s:%s:%s", shard, bucket.hashKey, metricName, v.Name),
+			RangeValue: encodeRangeKey(valueHash, seriesID, nil, labelSeriesRangeKeyV1),
+			Value:      []byte(v.Value),
+		})
+	}
+
+	return entries, nil
+}
+
+func (v11Entries) GetLabelNamesForSeries(bucket Bucket, seriesID []byte) ([]IndexQuery, error) {
+	return []IndexQuery{
+		{
+			TableName: bucket.tableName,
+			HashValue: string(seriesID),
+		},
+	}, nil
+}
diff --git a/pkg/chunk/schema_caching.go b/pkg/chunk/schema_caching.go
index 6db7105d93..73c8986269 100644
--- a/pkg/chunk/schema_caching.go
+++ b/pkg/chunk/schema_caching.go
@@ -46,6 +46,14 @@ func (s *schemaCaching) GetChunksForSeries(from, through model.Time, userID stri
 	return s.setImmutability(from, through, queries), nil
 }
 
+func (s *schemaCaching) GetLabelNamesForSeries(from, through model.Time, userID string, seriesID []byte) ([]IndexQuery, error) {
+	queries, err := s.Schema.GetLabelNamesForSeries(from, through, userID, seriesID)
+	if err != nil {
+		return nil, err
+	}
+	return s.setImmutability(from, through, queries), nil
+}
+
 func (s *schemaCaching) setImmutability(from, through model.Time, queries []IndexQuery) []IndexQuery {
 	cacheBefore := model.TimeFromUnix(mtime.Now().Add(-s.cacheOlderThan).Unix())
 
diff --git a/pkg/chunk/schema_config.go b/pkg/chunk/schema_config.go
index 1b630a615c..4e3e768339 100644
--- a/pkg/chunk/schema_config.go
+++ b/pkg/chunk/schema_config.go
@@ -219,8 +219,12 @@ func (cfg *SchemaConfig) ForEachAfter(t model.Time, f func(config *PeriodConfig)
 
 // CreateSchema returns the schema defined by the PeriodConfig
 func (cfg PeriodConfig) CreateSchema() Schema {
-	var e entries
+	rowShards := uint32(16)
+	if cfg.RowShards > 0 {
+		rowShards = cfg.RowShards
+	}
 
+	var e entries
 	switch cfg.Schema {
 	case "v1":
 		e = originalEntries{}
@@ -237,14 +241,15 @@ func (cfg PeriodConfig) CreateSchema() Schema {
 	case "v9":
 		e = v9Entries{}
 	case "v10":
-		rowShards := uint32(16)
-		if cfg.RowShards > 0 {
-			rowShards = cfg.RowShards
-		}
-
 		e = v10Entries{
 			rowShards: rowShards,
 		}
+	case "v11":
+		e = v11Entries{
+			v10Entries: v10Entries{
+				rowShards: rowShards,
+			},
+		}
 	default:
 		return nil
 	}
diff --git a/pkg/chunk/schema_test.go b/pkg/chunk/schema_test.go
index 18eae083ba..95cbb1c4fc 100644
--- a/pkg/chunk/schema_test.go
+++ b/pkg/chunk/schema_test.go
@@ -8,8 +8,10 @@ import (
 	"sort"
 	"testing"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/test"
 )
@@ -333,3 +335,56 @@ func TestSchemaRangeKey(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkEncodeLabelsJson(b *testing.B) {
+	decoded := &labels.Labels{}
+	lbs := labels.FromMap(map[string]string{
+		"foo":      "bar",
+		"fuzz":     "buzz",
+		"cluster":  "test",
+		"test":     "test1",
+		"instance": "cortex-01",
+		"bar":      "foo",
+		"version":  "0.1",
+	})
+	json := jsoniter.ConfigFastest
+	var data []byte
+	var err error
+	for n := 0; n < b.N; n++ {
+		data, err = json.Marshal(lbs)
+		if err != nil {
+			panic(err)
+		}
+		err = json.Unmarshal(data, decoded)
+		if err != nil {
+			panic(err)
+		}
+	}
+	b.Log("data size", len(data))
+	b.Log("decode", decoded)
+}
+
+func BenchmarkEncodeLabelsString(b *testing.B) {
+	var decoded labels.Labels
+	lbs := labels.FromMap(map[string]string{
+		"foo":      "bar",
+		"fuzz":     "buzz",
+		"cluster":  "test",
+		"test":     "test1",
+		"instance": "cortex-01",
+		"bar":      "foo",
+		"version":  "0.1",
+	})
+	var data []byte
+	var err error
+	for n := 0; n < b.N; n++ {
+		data = []byte(lbs.String())
+		decoded, err = promql.ParseMetric(string(data))
+		if err != nil {
+			panic(err)
+		}
+	}
+	b.Log("data size", len(data))
+	b.Log("decode", decoded)
+
+}
diff --git a/pkg/chunk/series_store.go b/pkg/chunk/series_store.go
index 1bfda046f1..bf1a9ae81c 100644
--- a/pkg/chunk/series_store.go
+++ b/pkg/chunk/series_store.go
@@ -4,8 +4,10 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sort"
 
 	"github.com/go-kit/kit/log/level"
+	jsoniter "github.com/json-iterator/go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
@@ -199,6 +201,25 @@ func (c *seriesStore) LabelNamesForMetricName(ctx context.Context, userID string
 	}
 	level.Debug(log).Log("series-ids", len(seriesIDs))
 
+	// Lookup the series in the index to get label names.
+	labelNames, err := c.lookupLabelNamesBySeries(ctx, from, through, userID, seriesIDs)
+	if err != nil {
+		// looking up metrics by series is not supported falling back on chunks
+		if err == ErrNotSupported {
+			return c.lookupLabelNamesByChunks(ctx, from, through, userID, seriesIDs)
+		}
+		level.Error(log).Log("msg", "lookupLabelNamesBySeries", "err", err)
+		return nil, err
+	}
+	level.Debug(log).Log("labelNames", len(labelNames))
+
+	return labelNames, nil
+}
+
+func (c *seriesStore) lookupLabelNamesByChunks(ctx context.Context, from, through model.Time, userID string, seriesIDs []string) ([]string, error) {
+	log, ctx := spanlogger.New(ctx, "SeriesStore.lookupLabelNamesByChunks")
+	defer log.Span.Finish()
+
 	// Lookup the series in the index to get the chunks.
 	chunkIDs, err := c.lookupChunksBySeries(ctx, from, through, userID, seriesIDs)
 	if err != nil {
@@ -228,7 +249,6 @@ func (c *seriesStore) LabelNamesForMetricName(ctx context.Context, userID string
 	}
 	return labelNamesFromChunks(allChunks), nil
 }
-
 func (c *seriesStore) lookupSeriesByMetricNameMatchers(ctx context.Context, from, through model.Time, userID, metricName string, matchers []*labels.Matcher) ([]string, error) {
 	log, ctx := spanlogger.New(ctx, "SeriesStore.lookupSeriesByMetricNameMatchers", "metricName", metricName, "matchers", len(matchers))
 	defer log.Span.Finish()
@@ -367,6 +387,44 @@ func (c *seriesStore) lookupChunksBySeries(ctx context.Context, from, through mo
 	return result, err
 }
 
+func (c *seriesStore) lookupLabelNamesBySeries(ctx context.Context, from, through model.Time, userID string, seriesIDs []string) ([]string, error) {
+	log, ctx := spanlogger.New(ctx, "SeriesStore.lookupLabelNamesBySeries")
+	defer log.Span.Finish()
+
+	level.Debug(log).Log("seriesIDs", len(seriesIDs))
+	queries := make([]IndexQuery, 0, len(seriesIDs))
+	for _, seriesID := range seriesIDs {
+		qs, err := c.schema.GetLabelNamesForSeries(from, through, userID, []byte(seriesID))
+		if err != nil {
+			return nil, err
+		}
+		queries = append(queries, qs...)
+	}
+	level.Debug(log).Log("queries", len(queries))
+	entries, err := c.lookupEntriesByQueries(ctx, queries)
+	if err != nil {
+		return nil, err
+	}
+	level.Debug(log).Log("entries", len(entries))
+	result := []string{model.MetricNameLabel}
+	uniqueLabelNames := map[string]struct{}{model.MetricNameLabel: {}}
+	for _, entry := range entries {
+		lbs := []string{}
+		err := jsoniter.ConfigFastest.Unmarshal(entry.Value, &lbs)
+		if err != nil {
+			return nil, err
+		}
+		for _, l := range lbs {
+			if _, ok := uniqueLabelNames[l]; !ok {
+				uniqueLabelNames[l] = struct{}{}
+				result = append(result, l)
+			}
+		}
+	}
+	sort.Strings(result)
+	return result, nil
+}
+
 // Put implements ChunkStore
 func (c *seriesStore) Put(ctx context.Context, chunks []Chunk) error {
 	for _, chunk := range chunks {