// Returns all (z, y, x0, x1) Spans in sorted order: z, then y, then x0.
func getSpans(ctx storage.VersionedCtx, minIndex, maxIndex indexRLE) ([]dvid.Span, error) {
	db, err := storage.SmallDataStore()
	if err != nil {
		return nil, err
	}
	spans := []dvid.Span{}

	var f storage.ChunkFunc = func(chunk *storage.Chunk) error {
		ibytes, err := chunk.K.ClassBytes(keyROI)
		if err != nil {
			return err
		}
		index := new(indexRLE)
		if err = index.IndexFromBytes(ibytes); err != nil {
			return fmt.Errorf("Unable to get indexRLE out of []byte encoding: %v\n", err)
		}
		z := index.start.Value(2)
		y := index.start.Value(1)
		x0 := index.start.Value(0)
		x1 := x0 + int32(index.span) - 1
		spans = append(spans, dvid.Span{z, y, x0, x1})
		return nil
	}
	mintk := storage.NewTKey(keyROI, minIndex.Bytes())
	maxtk := storage.NewTKey(keyROI, maxIndex.Bytes())
	err = db.ProcessRange(ctx, mintk, maxtk, &storage.ChunkOp{}, f)
	return spans, err
}

// NewTKey returns a TKey for storing a "label + spatial index", where
// the spatial index references a block that contains a voxel with the given label.
func NewTKey(label uint64, block dvid.IZYXString) storage.TKey {
	sz := len(block)
	ibytes := make([]byte, 8+sz)
	binary.BigEndian.PutUint64(ibytes[0:8], label)
	copy(ibytes[8:], []byte(block))
	return storage.NewTKey(keyLabelBlockRLE, ibytes)
}

// NewTagTKey returns a TKey for a given tag.
func NewTagTKey(tag Tag) (storage.TKey, error) {
	// Make sure the key has no embedded 0 values
	for i := 0; i < len(tag); i++ {
		if tag[i] == 0 {
			return nil, fmt.Errorf("tag cannot have embedded 0 value")
		}
	}
	return storage.NewTKey(keyTag, append([]byte(tag), 0)), nil
}

func (m *repoManager) putData(t storage.TKeyClass, data interface{}) error {
	var ctx storage.MetadataContext
	var buf bytes.Buffer
	enc := gob.NewEncoder(&buf)
	if err := enc.Encode(data); err != nil {
		return err
	}
	return m.store.Put(ctx, storage.NewTKey(t, nil), buf.Bytes())
}

// NewTypeSizeLabelTKey returns a type-specific key for the (index type, size, label) tuple.
func NewTypeSizeLabelTKey(i IndexType, sz uint32, label uint64) storage.TKey {
	// Since we want biggest -> smallest sort and initially only have forward range queries,
	// modify size to make smallest <-> largest.
	rsz := math.MaxUint32 - sz

	buf := make([]byte, 8+1+4)
	buf[0] = byte(i)
	binary.BigEndian.PutUint32(buf[1:5], rsz)
	binary.BigEndian.PutUint64(buf[5:], label)
	return storage.NewTKey(keyTypeSizeLabel, buf)
}

func (r *repoT) save() error {
	compression, err := dvid.NewCompression(dvid.LZ4, dvid.DefaultCompression)
	if err != nil {
		return err
	}
	serialization, err := dvid.Serialize(r, compression, dvid.CRC32)
	if err != nil {
		return err
	}

	var ctx storage.MetadataContext
	return manager.store.Put(ctx, storage.NewTKey(repoKey, r.id.Bytes()), serialization)
}

func (m *repoManager) loadData(t storage.TKeyClass, data interface{}) (found bool, err error) {
	var ctx storage.MetadataContext
	value, err := m.store.Get(ctx, storage.NewTKey(t, nil))
	if err != nil {
		return false, fmt.Errorf("Bad metadata GET: %v", err)
	}
	if value == nil {
		return false, nil
	}
	buf := bytes.NewBuffer(value)
	dec := gob.NewDecoder(buf)
	if err := dec.Decode(data); err != nil {
		return false, fmt.Errorf("Could not decode Gob encoded metadata (len %d): %v", len(value), err)
	}
	return true, nil
}

// Load the next ids to be used for RepoID, VersionID, and InstanceID.
func (m *repoManager) loadNewIDs() error {
	var ctx storage.MetadataContext
	value, err := m.store.Get(ctx, storage.NewTKey(newIDsKey, nil))
	if err != nil {
		return err
	}
	if len(value) != dvid.RepoIDSize+dvid.VersionIDSize+dvid.InstanceIDSize {
		return fmt.Errorf("Bad value returned for new ids.  Length %d bytes!", len(value))
	}
	pos := 0
	m.repoID = dvid.RepoIDFromBytes(value[pos : pos+dvid.RepoIDSize])
	pos += dvid.RepoIDSize
	m.versionID = dvid.VersionIDFromBytes(value[pos : pos+dvid.VersionIDSize])
	pos += dvid.VersionIDSize
	m.instanceID = dvid.InstanceIDFromBytes(value[pos : pos+dvid.InstanceIDSize])
	return nil
}

// PutSpans saves a slice of spans representing an ROI into the datastore.
// If the init parameter is true, all previous spans of this ROI are deleted before
// writing these spans.
func (d *Data) PutSpans(versionID dvid.VersionID, spans []dvid.Span, init bool) error {
	ctx := datastore.NewVersionedCtx(d, versionID)

	db, err := storage.SmallDataStore()
	if err != nil {
		return err
	}

	// Delete the old key/values
	if init {
		if err := d.Delete(ctx); err != nil {
			return err
		}
	}

	// Make sure our small data store can do batching.
	batcher, ok := db.(storage.KeyValueBatcher)
	if !ok {
		return fmt.Errorf("Unable to store ROI: small data store can't do batching!")
	}

	// We only want one PUT on given version for given data to prevent interleaved PUTs.
	putMutex := ctx.Mutex()
	putMutex.Lock()

	// Save new extents after finished.
	defer func() {
		err := datastore.SaveDataByVersion(ctx.VersionID(), d)
		if err != nil {
			dvid.Errorf("Error in trying to save repo on roi extent change: %v\n", err)
		}
		putMutex.Unlock()
	}()

	// Put the new key/values
	const BATCH_SIZE = 10000
	batch := batcher.NewBatch(ctx)
	for i, span := range spans {
		if span[0] < d.MinZ {
			d.MinZ = span[0]
		}
		if span[0] > d.MaxZ {
			d.MaxZ = span[0]
		}
		if span[3] < span[2] {
			return fmt.Errorf("Got weird span %v.  span[3] (X1) < span[2] (X0)", span)
		}
		index := indexRLE{
			start: dvid.IndexZYX{span[2], span[1], span[0]},
			span:  uint32(span[3] - span[2] + 1),
		}
		tk := storage.NewTKey(keyROI, index.Bytes())
		batch.Put(tk, dvid.EmptyValue())
		if (i+1)%BATCH_SIZE == 0 {
			if err := batch.Commit(); err != nil {
				return fmt.Errorf("Error on batch PUT at span %d: %v\n", i, err)
			}
			batch = batcher.NewBatch(ctx)
		}
	}
	if len(spans)%BATCH_SIZE != 0 {
		if err := batch.Commit(); err != nil {
			return fmt.Errorf("Error on last batch PUT: %v\n", err)
		}
	}
	return nil
}

// deletes a Repo from the datastore
func (r *repoT) delete() error {
	var ctx storage.MetadataContext
	tkey := storage.NewTKey(repoKey, r.id.Bytes())
	return manager.store.Delete(ctx, tkey)
}

func NewLabelTKey(label uint64) storage.TKey {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, label)
	return storage.NewTKey(keyLabel, buf)
}

// NewTypeLabelTKey returns a type-specific key for the (index type, label) tuple.
func NewTypeLabelTKey(i IndexType, label uint64) storage.TKey {
	buf := make([]byte, 1+8)
	buf[0] = byte(i)
	binary.BigEndian.PutUint64(buf[1:], label)
	return storage.NewTKey(keyTypeLabel, buf)
}

func NewBlockTKey(pt dvid.ChunkPoint3d) storage.TKey {
	idx := dvid.IndexZYX(pt)
	return storage.NewTKey(keyBlock, idx.Bytes())
}

// NewSizeLabelTKey returns a key component for storing a "size + label".
func NewSizeLabelTKey(size, label uint64) storage.TKey {
	ibytes := make([]byte, 16)
	binary.BigEndian.PutUint64(ibytes[0:8], size)
	binary.BigEndian.PutUint64(ibytes[8:16], label)
	return storage.NewTKey(keySizeLabel, ibytes)
}

// NewTKeyByCoord returns a TKey for a block coord in string format.
func NewTKeyByCoord(izyx dvid.IZYXString) storage.TKey {
	return storage.NewTKey(keyImageBlock, []byte(izyx))
}

// no checking of key
func newTKey(key string) storage.TKey {
	return storage.NewTKey(keyStandard, append([]byte(key), 0))
}

func (m *repoManager) putNewIDs() error {
	var ctx storage.MetadataContext
	value := append(m.repoID.Bytes(), m.versionID.Bytes()...)
	value = append(value, m.instanceID.Bytes()...)
	return m.store.Put(ctx, storage.NewTKey(newIDsKey, nil), value)
}

// NewLabelSizeTKey returns a key component for a "label + size".
func NewLabelSizeTKey(label, size uint64) storage.TKey {
	ibytes := make([]byte, 16)
	binary.BigEndian.PutUint64(ibytes[0:8], label)
	binary.BigEndian.PutUint64(ibytes[8:16], size)
	return storage.NewTKey(keyLabelSize, ibytes)
}

func (m *repoManager) loadVersion0() error {
	// Load the maps
	if _, err := m.loadData(repoToUUIDKey, &(m.repoToUUID)); err != nil {
		return fmt.Errorf("Error loading repo to UUID map: %s", err)
	}
	if _, err := m.loadData(versionToUUIDKey, &(m.versionToUUID)); err != nil {
		return fmt.Errorf("Error loading version to UUID map: %s", err)
	}
	if err := m.loadNewIDs(); err != nil {
		return fmt.Errorf("Error loading new local ids: %s", err)
	}

	// Generate the inverse UUID to VersionID mapping.
	for v, uuid := range m.versionToUUID {
		m.uuidToVersion[uuid] = v
	}

	// Load all the repo data
	var ctx storage.MetadataContext
	minRepo := dvid.RepoID(0)
	maxRepo := dvid.RepoID(dvid.MaxRepoID)

	minTKey := storage.NewTKey(repoKey, minRepo.Bytes())
	maxTKey := storage.NewTKey(repoKey, maxRepo.Bytes())
	kvList, err := m.store.GetRange(ctx, minTKey, maxTKey)
	if err != nil {
		return err
	}

	var saveCache bool
	for _, kv := range kvList {
		var saveRepo bool

		ibytes, err := kv.K.ClassBytes(repoKey)
		if err != nil {
			return err
		}
		repoID := dvid.RepoIDFromBytes(ibytes)

		// Load each repo
		_, found := m.repoToUUID[repoID]
		if !found {
			return fmt.Errorf("Retrieved repo with id %d that is not in map.  Corrupt DB?", repoID)
		}
		r := &repoT{
			log:        []string{},
			properties: make(map[string]interface{}),
			data:       make(map[dvid.InstanceName]DataService),
		}
		if err = dvid.Deserialize(kv.V, r); err != nil {
			return fmt.Errorf("Error gob decoding repo %d: %v", repoID, err)
		}

		// Cache all UUID from nodes into our high-level cache
		for v, node := range r.dag.nodes {
			uuid, found := m.versionToUUID[v]
			if !found {
				dvid.Errorf("Version id %d found in repo %s (id %d) not in cache map. Adding it...", v, r.uuid, r.id)
				m.versionToUUID[v] = node.uuid
				m.uuidToVersion[node.uuid] = v
				uuid = node.uuid
				saveCache = true
			}
			m.repos[uuid] = r
		}

		// Populate the instance id -> dataservice map.
		for _, dataservice := range r.data {
			m.iids[dataservice.InstanceID()] = dataservice
		}

		// Update the sync graph with all syncable data instances in this repo
		for _, dataservice := range r.data {
			syncedData, syncable := dataservice.(Syncer)
			if syncable {
				for _, name := range syncedData.SyncedNames() {
					r.addSyncGraph(syncedData.InitSync(name))
				}
			}
		}

		// Load any mutable properties for the data instances.
		for _, dataservice := range r.data {
			mutator, mutable := dataservice.(InstanceMutator)
			if mutable {
				saveRepo, err = mutator.LoadMutable(r.version, m.formatVersion, RepoFormatVersion)
				if err != nil {
					return err
				}
			}
		}

		// If updates had to be made, save the migrated repo metadata.
		if saveRepo {
			dvid.Infof("Re-saved repo with root %s due to migrations.\n", r.uuid)
			if err := r.save(); err != nil {
				return err
			}
		}
	}
//.........这里部分代码省略.........

// NewTKey returns the "key" key component.
func NewTKey(key string) (storage.TKey, error) {
	return storage.NewTKey(keyStandard, append([]byte(key), 0)), nil
}