Method reportBatchStat

memstore/batchstats.go:72–117

(batchIDs map[int]string)

}

func (batchStats *BatchStatsReporter) reportBatchStat(batchIDs map[int]string) {
	timer := utils.GetRootReporter().GetTimer(utils.BatchSizeReportTime).Start()
	defer timer.Stop()

	tables := batchStats.memStore.GetSchemas()

	for table, schema := range tables {
		shards := batchStats.shardOwner.GetOwnedShards()
		for _, shardID := range shards {
			shard, err := batchStats.memStore.GetTableShard(table, shardID)
			if err != nil || shard == nil {
				continue
			}

			for batchID, name := range batchIDs {
				if batchID < 0 {
					totalSize := 0
					liveBatchIDs, numRecordsInLastBatch := shard.LiveStore.GetBatchIDs()
					for i, liveBatchID := range liveBatchIDs {
						batch := shard.LiveStore.GetBatchForRead(liveBatchID)
						if batch == nil {
							continue
						}
						size := batch.Capacity
						batch.RUnlock()
						if i == len(batchIDs)-1 {
							size = numRecordsInLastBatch
						}
						totalSize += size
					}
					utils.GetReporter(table, shardID).GetChildGauge(map[string]string{"time": name}, utils.BatchSize).Update(float64(totalSize))
				} else {
					if !schema.Schema.IsFactTable {
						continue
					}
					version := shard.ArchiveStore.GetCurrentVersion()
					batch := version.RequestBatch(int32(batchID))
					size := batch.Size
					version.Users.Done()
					utils.GetReporter(table, shardID).GetChildGauge(map[string]string{"time": name}, utils.BatchSize).Update(float64(size))
				}
			}
			shard.Users.Done()
		}
	}
}