| 263 | } |
| 264 | |
| 265 | func (c *Compactor) finishBlock(ctx context.Context, block *streamingBlock, l log.Logger) error { |
| 266 | _, span := tracer.Start(ctx, "vparquet.compactor.finishBlock") |
| 267 | defer span.End() |
| 268 | |
| 269 | bytesFlushed, err := block.Complete() |
| 270 | if err != nil { |
| 271 | return fmt.Errorf("error completing block: %w", err) |
| 272 | } |
| 273 | |
| 274 | level.Info(l).Log("msg", "wrote compacted block", |
| 275 | "version", block.meta.Version, |
| 276 | "tenantID", block.meta.TenantID, |
| 277 | "blockID", block.meta.BlockID.String(), |
| 278 | "startTime", block.meta.StartTime.String(), |
| 279 | "endTime", block.meta.EndTime.String(), |
| 280 | "totalObjects", block.meta.TotalObjects, |
| 281 | "size", block.meta.Size_, |
| 282 | "compactionLevel", block.meta.CompactionLevel, |
| 283 | "totalRecords", block.meta.TotalObjects, |
| 284 | "bloomShardCount", block.meta.BloomShardCount, |
| 285 | "footerSize", block.meta.FooterSize, |
| 286 | "replicationFactor", block.meta.ReplicationFactor, |
| 287 | "dedicatedColumns", fmt.Sprintf("%+v", block.meta.DedicatedColumns), |
| 288 | ) |
| 289 | |
| 290 | span.AddEvent("wrote compacted block") |
| 291 | span.SetAttributes( |
| 292 | attribute.String("blockID", block.meta.BlockID.String()), |
| 293 | ) |
| 294 | |
| 295 | compactionLevel := int(block.meta.CompactionLevel) - 1 |
| 296 | if c.opts.BytesWritten != nil { |
| 297 | c.opts.BytesWritten(compactionLevel, bytesFlushed) |
| 298 | } |
| 299 | return nil |
| 300 | } |
| 301 | |
| 302 | type rowPool struct { |
| 303 | pool sync.Pool |