csdavidc opened a new issue, #15034:
URL: https://github.com/apache/pinot/issues/15034
I setup pino locally using:
`docker run -p 9000:9000 apachepinot/pinot:latest QuickStart -type
stream`
Then created a realtime table with ingestion aggregation,
with this configuration:
``` json
{
"REALTIME": {
"tableName": "minutlySales_REALTIME",
"tableType": "REALTIME",
"segmentsConfig": {
"replication": "1",
"retentionTimeUnit": "MINUTES",
"retentionTimeValue": "5",
"replicasPerPartition": "1",
"timeColumnName": "minutesSinceEpoch",
"minimizeDataMovement": false
},
"tenants": {
"broker": "DefaultTenant",
"server": "DefaultTenant",
"tagOverrideConfig": {}
},
"tableIndexConfig": {
"rangeIndexVersion": 2,
"autoGeneratedInvertedIndex": false,
"createInvertedIndexDuringSegmentGeneration": false,
"loadMode": "MMAP",
"noDictionaryColumns": [
"sales_count",
"total_sales"
],
"enableDefaultStarTree": false,
"enableDynamicStarTreeCreation": false,
"aggregateMetrics": false,
"nullHandlingEnabled": false,
"columnMajorSegmentBuilderEnabled": true,
"optimizeDictionary": false,
"optimizeDictionaryForMetrics": false,
"optimizeDictionaryType": false,
"noDictionarySizeRatioThreshold": 0.85
},
"metadata": {},
"quota": {},
"routing": {},
"query": {},
"ingestionConfig": {
"streamIngestionConfig": {
"streamConfigMaps": [
{
"streamType": "kafka",
"stream.kafka.topic.name": "dailySales",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:9092",
"realtime.segment.flush.threshold.time": "100",
"realtime.segment.flush.threshold.size": "5"
}
],
"columnMajorSegmentBuilderEnabled": true,
"trackFilteredMessageOffsets": false,
"pauselessConsumptionEnabled": false
},
"transformConfigs": [
{
"columnName": "minutesSinceEpoch",
"transformFunction": "toEpochMinutes(\"timestamp\")"
}
],
"aggregationConfigs": [
{
"columnName": "total_sales",
"aggregationFunction": "SUM(price)"
},
{
"columnName": "sales_count",
"aggregationFunction": "COUNT(*)"
}
],
"continueOnError": false,
"rowTimeValueCheck": false,
"segmentTimeValueCheck": true
},
"isDimTable": false
}
}
```
schema configuration:
``` json
{
"schemaName": "minutlySales",
"enableColumnBasedNullHandling": false,
"dimensionFieldSpecs": [
{
"name": "product_name",
"dataType": "STRING",
"fieldType": "DIMENSION",
"notNull": false
}
],
"metricFieldSpecs": [
{
"name": "sales_count",
"dataType": "LONG",
"fieldType": "METRIC",
"notNull": false
},
{
"name": "total_sales",
"dataType": "DOUBLE",
"fieldType": "METRIC",
"notNull": false
}
],
"dateTimeFieldSpecs": [
{
"name": "minutesSinceEpoch",
"dataType": "LONG",
"fieldType": "DATE_TIME",
"notNull": false,
"format": "1:MINUTES:EPOCH",
"granularity": "1:MINUTES"
}
]
}
```
segment.flush.threshold.size is 5 and retention period is 5 minute.
I streamed a lot of records into the table and records was created as
expected into multiple segments,
However segments whose endTime is much over due retention period are not
getting deleted from the table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
