func (d *driver) upsertPutFileRecords(pachClient *client.APIClient, file *pfs.File, newRecords *pfs.PutFileRecords) error {
	prefix, err := d.scratchFilePrefix(file)
	if err != nil {
		return err
	}

	ctx := pachClient.Ctx()
	_, err = col.NewSTM(ctx, d.etcdClient, func(stm col.STM) error {
		commitsCol := d.openCommits.ReadOnly(ctx)
		var commit pfs.Commit
		err := commitsCol.Get(file.Commit.ID, &commit)
		if err != nil {
			return err
		}
		// Dumb check to make sure the unmarshalled value exists (and matches the current ID)
		// to denote that the current commit is indeed open
		if commit.ID != file.Commit.ID {
			return fmt.Errorf("commit %v is not open", file.Commit.ID)
		}
		recordsCol := d.putFileRecords.ReadWrite(stm)
		var existingRecords pfs.PutFileRecords
		return recordsCol.Upsert(prefix, &existingRecords, func() error {
			if newRecords.Tombstone {
				existingRecords.Tombstone = true
				existingRecords.Records = nil
			}
			existingRecords.Split = newRecords.Split
			existingRecords.Records = append(existingRecords.Records, newRecords.Records...)
			existingRecords.Header = newRecords.Header
			existingRecords.Footer = newRecords.Footer
			return nil
		})
	})
	if err != nil {
		return err
	}

	return err
}
func (d *driver) upsertPutFileRecords(pachClient *client.APIClient, file *pfs.File, newRecords *pfs.PutFileRecords) error {
	prefix, err := d.scratchFilePrefix(file)
	if err != nil {
		return err
	}

	ctx := pachClient.Ctx()
	_, err = col.NewSTM(ctx, d.etcdClient, func(stm col.STM) error {
		commitsCol := d.openCommits.ReadOnly(ctx)
		var commit pfs.Commit
		err := commitsCol.Get(file.Commit.ID, &commit)
		if err != nil {
			return err
		}
		
		
		if commit.ID != file.Commit.ID {
			return fmt.Errorf("commit %v is not open", file.Commit.ID)
		}
		recordsCol := d.putFileRecords.ReadWrite(stm)
		var existingRecords pfs.PutFileRecords
		return recordsCol.Upsert(prefix, &existingRecords, func() error {
			if newRecords.Tombstone {
				existingRecords.Tombstone = true
				existingRecords.Records = nil
			}
			existingRecords.Split = newRecords.Split
			existingRecords.Records = append(existingRecords.Records, newRecords.Records...)
			existingRecords.Header = newRecords.Header
			existingRecords.Footer = newRecords.Footer
			return nil
		})
	})
	if err != nil {
		return err
	}

	return err
}
func ( d * driver ) upsertPutFileRecords ( pachClient * client . APIClient , file * pfs . File , newRecords * pfs . PutFileRecords ) error { prefix , err := d . scratchFilePrefix ( file ) if err != nil { return err } ctx := pachClient . Ctx ( ) _ , err = col . NewSTM ( ctx , d . etcdClient , func ( stm col . STM ) error { commitsCol := d . openCommits . ReadOnly ( ctx ) var commit pfs . Commit err := commitsCol . Get ( file . Commit . ID , & commit ) if err != nil { return err } // Dumb check to make sure the unmarshalled value exists (and matches the current ID) // to denote that the current commit is indeed open if commit . ID != file . Commit . ID { return fmt . Errorf ( " " , file . Commit . ID ) } recordsCol := d . putFileRecords . ReadWrite ( stm ) var existingRecords pfs . PutFileRecords return recordsCol . Upsert ( prefix , & existingRecords , func ( ) error { if newRecords . Tombstone { existingRecords . Tombstone = true existingRecords . Records = nil } existingRecords . Split = newRecords . Split existingRecords . Records = append ( existingRecords . Records , newRecords . Records ... ) existingRecords . Header = newRecords . Header existingRecords . Footer = newRecords . Footer return nil } ) } ) if err != nil { return err } return err }
// Put the tree into the blob store
// Only write the records to etcd if the commit does exist and is open.
// To check that a key exists in etcd, we assert that its CreateRevision
// is greater than zero.
Put the tree into the blob store Only write the records to etcd if the commit does exist and is open . To check that a key exists in etcd we assert that its CreateRevision is greater than zero .