Hi team, pls help and tell the solution. this is the blocker to my work 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 tail pinot server log and try to reload the segment

Thank so much @Xiang Fu. "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?