-
Notifications
You must be signed in to change notification settings - Fork 3.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Wire up drop series parsing #1629
Changes from 30 commits
fd9c19a
918c5c8
27d7f45
6370a4e
7cfc2a5
6d5be3d
8306057
c2daa18
4948b62
495cedb
0d6b761
697e972
5ab3731
9a6e29e
61352f7
6114c81
2752ada
5df9726
7d74cca
c575ee9
d35e2a9
da10fa8
b53d02f
d5ac049
63b6719
3e94c14
d059b3e
113fcea
29910c3
8637a10
c8afd62
0f42be3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -221,6 +221,54 @@ func (m *Measurement) addSeries(s *Series) bool { | |
return true | ||
} | ||
|
||
// removeSeries will remove a series from the measurementIndex. Returns true if already removed | ||
func (m *Measurement) dropSeries(seriesID uint32) bool { | ||
if _, ok := m.seriesByID[seriesID]; !ok { | ||
return true | ||
} | ||
s := m.seriesByID[seriesID] | ||
tagset := string(marshalTags(s.Tags)) | ||
|
||
delete(m.series, tagset) | ||
delete(m.seriesByID, seriesID) | ||
|
||
var ids []uint32 | ||
for _, id := range m.seriesIDs { | ||
if id != seriesID { | ||
ids = append(ids, id) | ||
} | ||
} | ||
m.seriesIDs = ids | ||
|
||
// remove this series id to the tag index on the measurement | ||
// s.seriesByTagKeyValue is defined as map[string]map[string]seriesIDs | ||
for k, v := range m.seriesByTagKeyValue { | ||
values := v | ||
for kk, vv := range values { | ||
var ids []uint32 | ||
for _, id := range vv { | ||
if id != seriesID { | ||
ids = append(ids, id) | ||
} | ||
} | ||
// Check to see if we have any ids, if not, remove the key | ||
if len(ids) == 0 { | ||
delete(values, kk) | ||
} else { | ||
values[kk] = ids | ||
} | ||
} | ||
// If we have no values, then we delete the key | ||
if len(values) == 0 { | ||
delete(m.seriesByTagKeyValue, k) | ||
} else { | ||
m.seriesByTagKeyValue[k] = values | ||
} | ||
} | ||
|
||
return true | ||
} | ||
|
||
// seriesByTags returns the Series that matches the given tagset. | ||
func (m *Measurement) seriesByTags(tags map[string]string) *Series { | ||
return m.series[string(marshalTags(tags))] | ||
|
@@ -1002,6 +1050,17 @@ func (rp *RetentionPolicy) shardGroupByID(shardID uint64) *ShardGroup { | |
return nil | ||
} | ||
|
||
// dropSeries will delete all data with the seriesID | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why is this done by Retention Policy? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. My understanding is that is the only way to walk through to get the shards to delete the series. Is there a faster way to isolate the shards I need? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. no, retention policy is one of the right places to have it. If you're dropping a series, you should go from database -> retention policy -> shard group -> shard. Each of those will have a method |
||
func (rp *RetentionPolicy) dropSeries(seriesID uint32) error { | ||
for _, g := range rp.shardGroups { | ||
err := g.dropSeries(seriesID) | ||
if err != nil { | ||
return err | ||
} | ||
} | ||
return nil | ||
} | ||
|
||
func (rp *RetentionPolicy) removeShardGroupByID(shardID uint64) { | ||
for i, g := range rp.shardGroups { | ||
if g.ID == shardID { | ||
|
@@ -1075,6 +1134,35 @@ func (db *database) addSeriesToIndex(measurementName string, s *Series) bool { | |
return idx.addSeries(s) | ||
} | ||
|
||
// dropSeries removes the series from the in memory references | ||
func (db *database) dropSeries(seriesByMeasurement map[string][]uint32) error { | ||
for measurement, ids := range seriesByMeasurement { | ||
for _, id := range ids { | ||
// if the series is already gone, return | ||
if db.series[id] == nil { | ||
continue | ||
} | ||
|
||
delete(db.series, id) | ||
|
||
// Remove series information from measurements | ||
m := db.measurements[measurement] | ||
if !m.dropSeries(id) { | ||
return fmt.Errorf("failed to remove series id %d from measurment %q", id, m.Name) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This shouldn't throw an error. If it was already removed, we're ok with that, just |
||
} | ||
|
||
// Remove shard data | ||
for _, rp := range db.policies { | ||
if err := rp.dropSeries(id); err != nil { | ||
return err | ||
} | ||
} | ||
} | ||
} | ||
|
||
return nil | ||
} | ||
|
||
// createMeasurementIfNotExists will either add a measurement object to the index or return the existing one. | ||
func (db *database) createMeasurementIfNotExists(name string) *Measurement { | ||
idx := db.measurements[name] | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -859,14 +859,31 @@ func (p *Parser) parseShowFieldKeysStatement() (*ShowFieldKeysStatement, error) | |
// This function assumes the "DROP SERIES" tokens have already been consumed. | ||
func (p *Parser) parseDropSeriesStatement() (*DropSeriesStatement, error) { | ||
stmt := &DropSeriesStatement{} | ||
var err error | ||
|
||
// Read the name of the series to drop. | ||
lit, err := p.parseIdent() | ||
if err != nil { | ||
if tok, _, _ := p.scanIgnoreWhitespace(); tok == FROM { | ||
// Parse source. | ||
if stmt.Source, err = p.parseSource(); err != nil { | ||
return nil, err | ||
} | ||
} else { | ||
p.unscan() | ||
} | ||
|
||
// Parse condition: "WHERE EXPR". | ||
if stmt.Condition, err = p.parseCondition(); err != nil { | ||
return nil, err | ||
} | ||
stmt.Name = lit | ||
|
||
// If they didn't provide a FROM or a WHERE, they need to provide the SeriesID | ||
if stmt.Condition == nil && stmt.Source == nil { | ||
var id int | ||
id, err = p.parseInt(0, math.MaxUint32) | ||
if err != nil { | ||
return nil, err | ||
} | ||
stmt.SeriesID = uint32(id) | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I should have thought of this earlier when we were discussing syntax but it needs to support the syntax I mentioned in the comment above, on the |
||
return stmt, nil | ||
} | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can these two lookups of
seriesByID
not be combined? I think it would be more pleasing if there were.