This message was deleted.

Here is the ingestion spec:

"type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "radio_node_rollup",
      "timestampSpec": {
        "column": "start_time",
        "format": "millis",
        "missingValue": null
      },
      "dimensionsSpec": {
        "dimensions": [
          {
            "type": "string",
            "name": "enodeb_id_range_prefix",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "site_id",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "cell_id",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "record_type",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "call_type",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "call_direction",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "ca_status",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "cause_for_record_closure",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "service_reason_return_code",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          }
        ],
        "dimensionExclusions": [
          "duration",
          "__time",
          "start_time",
          "count",
          "sum_duration"
        ],
        "includeAllDimensions": false
      },
      "metricsSpec": [
        {
          "type": "count",
          "name": "count"
        },
        {
          "type": "longSum",
          "name": "sum_duration",
          "fieldName": "duration"
        }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "HOUR",
        "rollup": true,
        "intervals": []
      },
      "transformSpec": {
        "filter": null,
        "transforms": [
          {
            "type": "expression",
            "name": "enodeb_id_range_prefix",
            "expression": "substring(site_id,0,3)"
          }
        ]
      }
    },
    "ioConfig": {
      "topic": "raw-to-baseline-processed-data",
      "inputFormat": {
        "type": "json",
        "keepNullColumns": false,
        "assumeNewlineDelimited": false,
        "useJsonNodeReader": false
      },
      "replicas": 1,
      "taskCount": 8,
      "taskDuration": "PT3600S",
      "consumerProperties": {
        "bootstrap.servers": "xxx"
      },
      "autoScalerConfig": null,
      "pollTimeout": 100,
      "startDelay": "PT5S",
      "period": "PT30S",
      "useEarliestOffset": false,
      "completionTimeout": "PT5400S",
      "lateMessageRejectionPeriod": null,
      "earlyMessageRejectionPeriod": null,
      "lateMessageRejectionStartDateTime": null,
      "configOverrides": null,
      "idleConfig": null,
      "stream": "raw-to-baseline-processed-data",
      "useEarliestSequenceNumber": false,

      "type": "kafka"
    },
    "tuningConfig": {
      "type": "kafka",
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 1000000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxRowsPerSegment": 1000000,
      "maxTotalRows": null,
      "intermediatePersistPeriod": "PT10M",
      "maxPendingPersists": 0,
      "indexSpec": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "indexSpecForIntermediatePersists": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "reportParseExceptions": false,
      "handoffConditionTimeout": 0,
      "resetOffsetAutomatically": false,
      "segmentWriteOutMediumFactory": null,
      "workerThreads": null,
      "chatThreads": null,
      "chatRetries": 8,
      "httpTimeout": "PT10S",
      "shutdownTimeout": "PT80S",
      "offsetFetchPeriod": "PT30S",
      "intermediateHandoffPeriod": "P2147483647D",
      "logParseExceptions": true,
      "maxParseExceptions": 2147483647,
      "maxSavedParseExceptions": 0,
      "skipSequenceNumberAvailabilityCheck": false,
      "repartitionTransitionDuration": "PT120S"
    }
  },
  "context": null

Do you mean something like count(*), or sum(count)? If the former, I suspect that things get further rolled up when realtime segments are coalesced and persisted to deep storage.

we do count(*).

I think things are partially rolled up in (smaller) realtime segments, then further rolled up when persisted to durable segments.

@Ben Krug got it. could you please detail why this can happen please. ?

+1 to Ben's info --

count(*)

is the number of physical data records that exist. In a non-rollup table records are not created or removed, they are each distinct and stay distinct. So count(*) should always give you a consistent number. A rollup table, however, behaves like a Group By query ... which aggregates records together resulting in fewer records. Generally in a rollup table you always end up with fewer records than the number of events that were ingested, because the original event records are grouped together as part of the rollup process. Because ingestion does many things in parallel, you initially will have multiple ingested sets of data ... since they do not know about each other, they are only grouping within the data that they have ingested. As time progresses and these record sets are consolidated, grouping is done again, consolidating more records together and further reducing the physical record count. Because the number of physical records reduces as rollups occur, the metric column "count" may be created to record the original number of event records that came in. This is the

sum(count)

that Ben is referring to. Granted the metric name "count" is easily confused with the count(*) aggregate function, but they are different things. Doc ref: https://druid.apache.org/docs/latest/ingestion/rollup