Hi team, I’m trying to use Pinot upsert feature. Part of my table config is like below: { “tableName”: “upsert_test_local”, “tableType”: “REALTIME”, “segmentsConfig”: { “schemaName”: “upsert_test_local”, “timeColumnName”: “*created_on*”, “timeType”: “MILLISECONDS”, “allowNullTimeValue”: true, “replicasPerPartition”: “1", “retentionTimeUnit”: “DAYS”, “retentionTimeValue”: “30", “segmentPushType”: “APPEND”, “completionConfig”: { “completionMode”: “DOWNLOAD” } }, “tenants”: { }, “tableIndexConfig”: { “loadMode”: “MMAP”, “aggregateMetrics”: true, “nullHandlingEnabled”: true, “streamConfigs”: { “streamType”: “kafka”, “stream.kafka.consumer.type”: “lowlevel”, “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”: “largest”, “realtime.segment.flush.threshold.time”: “*30m*”, “realtime.segment.flush.threshold.rows”: “0", “realtime.segment.flush.threshold.segment.size”: “100M”, “realtime.segment.flush.autotune.initialRows”: “1000000" } }, “ingestionConfig”: { “filterConfig”: { “filterFunction”: “Groovy({tablename != \“test_table_name\“}, tablename)” }, “transformConfigs”: [ { “columnName”: “id”, “transformFunction”: “Groovy({UUID.randomUUID().toString()}, tablename)” }, { “columnName”: “*timestamp*”, “transformFunction”: “jsonPathString(metrics, ‘$.timestamp’)” }, { “columnName”: “*created_on*”, “transformFunction”: “Groovy({System.currentTimeMillis()}, tablename)” }, { “columnName”: “updated_on”, “transformFunction”: “Groovy({System.currentTimeMillis()}, tablename)” } ] }, “metadata”: { “customConfigs”: {} }, “routing”: { “instanceSelectorType”: “strictReplicaGroup” }, “upsertConfig”: { “mode”: “PARTIAL”, “defaultPartialUpsertStrategy”: “OVERWRITE”, “partialUpsertStrategies”:{ “*created_on*”: “IGNORE” } } } And part of schema is like this: “dateTimeFieldSpecs”: [ { “name”: “*timestamp*”, “dataType”: “LONG”, “format”: “1MILLISECONDSEPOCH”, “granularity”: “1:MILLISECONDS” }, { “name”: “*created_on*”, “dataType”: “LONG”, “format”: “1MILLISECONDSEPOCH”, “granularity”: “1:MILLISECONDS” }, { “name”: “updated_on”, “dataType”: “LONG”, “format”: “1MILLISECONDSEPOCH”, “granularity”: “1:MILLISECONDS” } ], “primaryKeyColumns”: [ “timestamp” ] At first, upsert works as expected. But after a while, like 30 minutes later, when I query this table, there is no record in this table. But totalDocs in the query response stats is not 0. Then I write some data to the same Kafka topic, and query this table, there are some records. But the value of the created_on field is 0 instead of the current timestamp. Any idea what property is not set right here? Is it timeColumnName property?

Hi, I see you are using timestamp column as primaryKey, is that expected? Ideally for upsert to work your primaryKey should be unique for a record and your input kafka stream should be partitioned on the primary key.

Hi, I’ve replicated this error. The records are there(use skipupsert option). But can’t be queried.

Hi, when you apply

skipUpsert

option in query, do you get both old data + updated data or just old data?

And when I add another 2 records to the same Kafka topic. I think it’s ingested to Pinot from the info in the following picture. The only problem is the created_on is 0 here. It should be a long type timestamp.

How many servers do you have? Since it returns 2 records with the same timestamp, I suspect the source steam is not partitioned on the primary key

Hi, only one Kafka topic partition and 6 servers.

So I've tried reproducing this issue on both 0.10 and master build. The

No records(s) Found

error does occur in both. Need to check why is that the case. The

created_on

as 0 is not reproducible so far. It shows correct values in all tried out permutations.

@Kartik Khare Great analysis! IMO we should not allow modifying of the

comparisonColumn

. We use

comparisonColumn

to determine which record to keep, and allow modifying it can break the contract

Yep. So I guess then we should add a validation config in table so that it warns user

We can add the validation in

TableConfigUtils.validateUpsertConfig()

to reject table config with upsert strategy on primary key columns or comparison column. In

PartialUpsertHandler

, we may log warning if we find strategy configured for these columns, and don't add them to the mergers in order to fix the tables already exist

cool. will raise the fix