Hello everyone! I am not 100% sure if my understanding for the merge rollup task is correct and I am looking for some clarification. I have the following table and schema config:

{
  "REALTIME": {
    "tableName": "TestRollup_REALTIME",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "schemaName": "TestRollup",
      "replication": "1",
      "replicasPerPartition": "1",
      "timeColumnName": "ValueTimestamp",
      "minimizeDataMovement": false
    },
    "tenants": {
      "broker": "DefaultTenant",
      "server": "DefaultTenant",
      "tagOverrideConfig": {}
    },
    "tableIndexConfig": {
      "invertedIndexColumns": [],
      "noDictionaryColumns": [],
      "streamConfigs": {
        "streamType": "kafka",
        "stream.kafka.topic.name": "rollup-price-aggregation",
        "stream.kafka.broker.list": "kafka:9092",
        "stream.kafka.consumer.type": "lowlevel",
        "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
        "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
        "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
        "realtime.segment.flush.threshold.rows": "0",
        "realtime.segment.flush.threshold.segment.rows": "500",
        "realtime.segment.flush.threshold.time": "20m"
      },
      "aggregateMetrics": false,
      "enableDefaultStarTree": false,
      "nullHandlingEnabled": false,
      "bloomFilterColumns": [],
      "onHeapDictionaryColumns": [],
      "rangeIndexColumns": [],
      "sortedColumn": [],
      "varLengthDictionaryColumns": [],
      "rangeIndexVersion": 2,
      "optimizeDictionaryForMetrics": false,
      "optimizeDictionary": false,
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "loadMode": "MMAP",
      "enableDynamicStarTreeCreation": false,
      "columnMajorSegmentBuilderEnabled": true,
      "optimizeDictionaryType": false,
      "noDictionarySizeRatioThreshold": 0.85
    },
    "metadata": {},
    "quota": {},
    "task": {
      "taskTypeConfigsMap": {
        "MergeRollupTask": {
          "5m_2m.mergeType": "rollup",
          "5m_2m.bucketTimePeriod": "5m",
          "5m_2m.bufferTimePeriod": "2m",
          "5m_2m.roundBucketTimePeriod": "1m",
          "10m_2m.mergeType": "rollup",
          "10m_2m.bucketTimePeriod": "10m",
          "10m_2m.bufferTimePeriod": "12m",
          "10m_2m.roundBucketTimePeriod": "5m",
          "Price.aggregationType": "sum",
          "schedule": "0 * * * * ?"
        }
      }
    },
    "routing": {},
    "query": {},
    "ingestionConfig": {
      "continueOnError": false,
      "rowTimeValueCheck": false,
      "segmentTimeValueCheck": true
    },
    "isDimTable": false
  }
}

{
  "schemaName": "TestRollup",
  "enableColumnBasedNullHandling": false,
  "dimensionFieldSpecs": [
    {
      "name": "Id",
      "dataType": "INT",
      "fieldType": "DIMENSION",
      "notNull": false
    }
  ],
  "metricFieldSpecs": [
    {
      "name": "Price",
      "dataType": "DOUBLE",
      "fieldType": "METRIC",
      "notNull": false
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ValueTimestamp",
      "dataType": "TIMESTAMP",
      "fieldType": "DATE_TIME",
      "notNull": false,
      "format": "TIMESTAMP",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

I am ingesting data from a Kafka topic and all the records have a ValueTimestamp equal to now. From my understanding, with that configuration, the merge rollup tasks in the config will be executed once a minute and create merged segments based on the provided configuration - if there are records whose timestamp is within the timeframe of (bucketStart -> bucketStart + 5m or bucketStart -> bucketStart + 10m) and are older than 2m or older than 12m they will be added to the merged segment, is that correct? Also, for the rounding of the timestamp, the records in the merged segment will have timestamps that are rounded to 1m or rounded to 5m (for example: they will have timestamps like 2025-04-11 1150/1151 or 2025-04-11 1150/1155)?

What do you want to achieve with the task? once a minute is too small an interval, it should be order of hours.

It is for testing purposes of just seeing how the task works in practice. For real data I will use intervals in hours/days.

From my understanding, with that configuration, the merge rollup tasks in the config will be executed once a minute and create merged segments based on the provided configuration

This understanding is correct. However, this is a very aggressive schedule practically as noted by Mayank.

if there are records whose timestamp is within the timeframe of (bucketStart -> bucketStart + 5m or bucketStart -> bucketStart + 10m) and are older than 2m or older than 12m they will be added to the merged segment, is that correct?

This understanding is correct.

Also, for the rounding of the timestamp, the records in the merged segment will have timestamps that are rounded to 1m or rounded to 5m (for example: they will have timestamps like 2025-04-11 1150/1151 or 2025-04-11 1150/1155)?

This understanding is correct.

Thanks @Ragesh Rajagopalan

🍻Thanks everyone!