Hello, everyone! I currently have this set up for Kafka ingestion in Pinot - I ingest data from a Kafka topic of the form Id - INT, Price - DOUBLE, UnixTime - LONG into a Pinot REALTIME table. I wanted to ask if there is a way to use the UnixTime timestamp from the ingested data as the timestamp column in the table - to use the value in the ingested message as Timestamp instead of ingestion time?

can you drop your schema and table config here?

This is currently what I am testing: { "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" }, { "name": "Timestamp", "dataType": "TIMESTAMP", "fieldType": "DATE_TIME", "notNull": false, "format": "TIMESTAMP", "granularity": "1:MILLISECONDS" } ] } { "REALTIME": { "tableName": "TestRollup_REALTIME", "tableType": "REALTIME", "segmentsConfig": { "schemaName": "TestRollup", "replication": "1", "replicasPerPartition": "1", "timeColumnName": "Timestamp", "minimizeDataMovement": false }, "tenants": { "broker": "DefaultTenant", "server": "DefaultTenant", "tagOverrideConfig": {} }, "tableIndexConfig": { "enableDefaultStarTree": false, "nullHandlingEnabled": false, "noDictionaryColumns": [], "bloomFilterColumns": [], "invertedIndexColumns": [], "onHeapDictionaryColumns": [], "rangeIndexColumns": [], "sortedColumn": [], "varLengthDictionaryColumns": [], "rangeIndexVersion": 2, "optimizeDictionaryForMetrics": false, "optimizeDictionary": false, "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": "50", "realtime.segment.flush.threshold.time": "5m" }, "aggregateMetrics": false, "optimizeDictionaryType": false, "noDictionarySizeRatioThreshold": 0.85, "autoGeneratedInvertedIndex": false, "createInvertedIndexDuringSegmentGeneration": false, "loadMode": "MMAP", "enableDynamicStarTreeCreation": false, "columnMajorSegmentBuilderEnabled": true }, "metadata": {}, "quota": {}, "routing": {}, "query": {}, "ingestionConfig": { "rowTimeValueCheck": false, "continueOnError": false, "segmentTimeValueCheck": true }, "MergeRollupTask": { "1hour.mergeType": "rollup", "1hour.bucketTimePeriod": "1h", "1hour.bufferTimePeriod": "1h", "Price.aggregationType": "sum", }, "isDimTable": false } } I have a value in my message in Kafka (ValueTimestamp) and Timestamp is what I've set as Timestamp column in my REALTIME table. In that case, the Timestamp column will contain the ingestion date, right?

why you need both

ValueTimestamp

and

Timestamp

?

Yes, my bad, I've copied the config wrong it is with the correct configuration.

ack

Yes, but when I choose "ValueTimestamp" as my timeColumnName, the values in that column are not the ones from the messages in Kafka but the time of ingestion in Pinot (I assume)

can you confirm? Pinot doesn’t use ingestion time

I will run some tests to make sure but I think I made a mistake in the table config.