Hi team, pls help and tell the solution guys. this is the blocker to my work and let me know production level suggestions. I added the upsert configuration to the existing setup, but it didn’t work initially. After deleting the old configuration and re-adding the same setup with upserts, it started working. However, after some time, the segments began failing. { "REALTIME": { "tableName": "views_REALTIME", "tableType": "REALTIME", "segmentsConfig": { "schemaName": "views", "replication": "1", "retentionTimeUnit": "DAYS", "retentionTimeValue": "90", "replicasPerPartition": "1", "timeColumnName": "view_end", "minimizeDataMovement": false }, "tenants": { "broker": "DefaultTenant", "server": "DefaultTenant", "tagOverrideConfig": {} }, "tableIndexConfig": { "aggregateMetrics": false, "starTreeIndexConfigs": [], "enableDefaultStarTree": false, "nullHandlingEnabled": false, "noDictionaryColumns": [ "events" ], "invertedIndexColumns": [ "workspace_id", "country_code", "fp_playback_id", "browser_name", "is_final" ], "bloomFilterColumns": [], "onHeapDictionaryColumns": [], "rangeIndexColumns": [ "view_end", "view_start", "created_at" ], "sortedColumn": [ "view_end", "quality_of_experience_score", "playback_score", "render_quality_score", "stability_score", "startup_score" ], "varLengthDictionaryColumns": [], "rangeIndexVersion": 2, "optimizeDictionaryForMetrics": false, "optimizeDictionary": false, "autoGeneratedInvertedIndex": false, "createInvertedIndexDuringSegmentGeneration": false, "loadMode": "MMAP", "enableDynamicStarTreeCreation": true, "columnMajorSegmentBuilderEnabled": true, "noDictionarySizeRatioThreshold": 0.85 }, "metadata": {}, "quota": {}, "task": { "taskTypeConfigsMap": { "UpsertCompactionTask": { "schedule": "0 0 * ? * *", "bufferTimePeriod": "1h", "invalidRecordsThresholdPercent": "30", "invalidRecordsThresholdCount": "100000", "tableMaxNumTasks": "10", "validDocIdsType": "SNAPSHOT" } } }, "routing": { "segmentPrunerTypes": [ "partition" ], "instanceSelectorType": "strictReplicaGroup" }, "query": {}, "upsertConfig": { "enableSnapshot": true, "deletedKeysTTL": 0, "mode": "FULL", "comparisonColumns": [ "view_end" ], "metadataTTL": 0, "dropOutOfOrderRecord": false, "hashFunction": "NONE", "defaultPartialUpsertStrategy": "OVERWRITE", "enablePreload": true, "consistencyMode": "NONE", "upsertViewRefreshIntervalMs": 3000, "allowPartialUpsertConsumptionDuringCommit": false }, "ingestionConfig": { "transformConfigs": [ { "columnName": "created_at", "transformFunction": "Now()" } ], "streamIngestionConfig": { "streamConfigMaps": [ { "streamType": "kafka", "stream.kafka.topic.name": "fp-data-processed-views-v1", "stream.kafka.consumer.prop.group.id": "pinot-views", "stream.kafka.broker.list": "kafka-cluster-broker-0.kafka-cluster-kafka-brokers.prod-kafka.svc.cluster.local9092,kafka cluster broker 1.kafka cluster kafka brokers.prod kafka.svc.cluster.local9092,kafka-cluster-broker-2.kafka-cluster-kafka-brokers.prod-kafka.svc.cluster.local:9092", "stream.kafka.consumer.type": "lowlevel", "stream.kafka.consumer.prop.auto.offset.reset": "largest", "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", "sasl.mechanism": "SCRAM-SHA-512", "security.protocol": "SASL_PLAINTEXT", "sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"hfjcnfrjnrc\" password=\"njffhirjfkriviruuir\";", "realtime.segment.flush.threshold.rows": "0", "realtime.segment.flush.threshold.segment.size": "200M", "realtime.segment.flush.threshold.time": "24h" } ], "columnMajorSegmentBuilderEnabled": true, "trackFilteredMessageOffsets": false }, "continueOnError": false, "rowTimeValueCheck": false, "segmentTimeValueCheck": true }, "isDimTable": false } }

Can you hover on the BAD status to see if it shows the error? If not, use debug api

Does it mean we would have to create a new table for UPSERT? and then migrate traffic over?

Yes

that is a non-trivial effort

Upsert doesn’t work with offline table. Are you expecting that already existing data will be upserted?

yea I knew upsert doesn’t work with offline but can I have upsert for realtime and also have a offline table?

Ok, then you can simply enable upsert. But if you have a PK that already existed before enabling you will see multiple records

gotcha - would i need to restart servers after the config changes?

Here’s a clear and professional rephrase: Hello @Mayank, I enabled upserts on an existing table, but it didn’t work initially. After restarting the Pinot server and broker, it started working. It’s working fine now, but I want to confirm whether a restart is required at the production level. This will be going to production in a few days, so I want to be sure.

Yes

Thank so much @Mayank. "task": { "taskTypeConfigsMap": { "UpsertCompactionTask": { "schedule": "0 0 * ? * *", "bufferTimePeriod": "1h", "invalidRecordsThresholdPercent": "30", "invalidRecordsThresholdCount": "100000", "tableMaxNumTasks": "10", "validDocIdsType": "SNAPSHOT" } } }, "routing": { "segmentPrunerTypes": [ "partition" ], "instanceSelectorType": "strictReplicaGroup" }, "query": {}, "upsertConfig": { "dropOutOfOrderRecord": false, "mode": "FULL", "consistencyMode": "NONE", "metadataTTL": 0, "hashFunction": "NONE", "defaultPartialUpsertStrategy": "OVERWRITE", "upsertViewRefreshIntervalMs": 3000, "enableSnapshot": true, "deletedKeysTTL": 0, "enablePreload": true, "newSegmentTrackingTimeMs": 10000, "enableDeletedKeysCompactionConsistency": false, "allowPartialUpsertConsumptionDuringCommit": false }, This is my upsert configuration based on the documentation. Can you review it and let me know whether it's suitable for production? I’ve also enabled upsert compaction, but it hasn’t started yet. Can you tell me why it’s still in that state?