So, I'm trying to configure a table with partial upsert on

Pinot 0.9.0

that consumes from a kafka topic and I'm experiencing a weird behaviour. Once the second message gets consumed, Pinot does a full upsert instead of the partial. So every field present in the second message gets updated and all the others are set to null (I believe because they are not present on the second message and the full upsert uses the default null values) Here's the table and schema configs: Schema:

{
  "schemaName": "responseCount",
  "dimensionFieldSpecs": [
    {
      "name": "responseId",
      "dataType": "STRING"
    },
    {
      "name": "formId",
      "dataType": "STRING"
    },
    {
      "name": "channelId",
      "dataType": "STRING"
    },
    {
      "name": "channelPlatform",
      "dataType": "STRING"
    },
    {
      "name": "companyId",
      "dataType": "STRING"
    },
    {
      "name": "submitted",
      "dataType": "BOOLEAN"
    },
    {
      "name": "deleted",
      "dataType": "BOOLEAN"
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "operationDate",
      "dataType": "STRING",
      "format": "1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd'T'HH:mm:ss.SSSZ",
      "granularity": "1:MILLISECONDS"
    },
    {
      "name": "createdAt",
      "dataType": "STRING",
      "format": "1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd'T'HH:mm:ss.SSSZ",
      "granularity": "1:MILLISECONDS"
    },
    {
      "name": "deletedAt",
      "dataType": "STRING",
      "format": "1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd'T'HH:mm:ss.SSSZ",
      "granularity": "1:MILLISECONDS"
    }
  ],
  "primaryKeyColumns": [
    "responseId"
  ]
}

Table:

{
  "REALTIME": {
    "tableName": "responseCount_REALTIME",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "allowNullTimeValue": false,
      "replication": "1",
      "replicasPerPartition": "1",
      "timeColumnName": "operationDate",
      "schemaName": "responseCount"
    },
    "tenants": {
      "broker": "DefaultTenant",
      "server": "DefaultTenant"
    },
    "tableIndexConfig": {
      "rangeIndexVersion": 1,
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "loadMode": "MMAP",
      "streamConfigs": {
        "streamType": "kafka",
        "stream.kafka.topic.name": "response-count.aggregation.source",
        "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.time": "24h",
        "realtime.segment.flush.segment.size": "100M"
      },
      "enableDefaultStarTree": false,
      "enableDynamicStarTreeCreation": false,
      "aggregateMetrics": false,
      "nullHandlingEnabled": true
    },
    "metadata": {},
    "routing": {
      "instanceSelectorType": "strictReplicaGroup"
    },
    "upsertConfig": {
      "mode": "PARTIAL",
      "partialUpsertStrategies": {
        "deleted": "OVERWRITE",
        "deletedAt": "OVERWRITE"
      },
      "hashFunction": "NONE"
    },
    "isDimTable": false
  }
}

Here's the first message consumed:

Key: {"responseId": "52d96a0d-92ea-4103-9ea9-536252324481"}
Value:
{
  "responseId": "52d96a0d-92ea-4103-9ea9-536252324481",
  "formId": "7bd28941-f9e4-45f1-a801-5c7d647cc6cd",
  "channelId": "60d11312-0e01-48d8-acce-4871b8d2365b",
  "channelPlatform": "app",
  "companyId": "00ca0142-5634-57e6-8d44-61427ea4b13d",
  "submitted": true,
  "deleted": "false",
  "createdAt": "2021-05-21T12:55:54.000+0000",
  "operationDate": "2021-05-21T12:55:54.000+0000"
}

Here's the second message consumed:

Key: {"responseId": "52d96a0d-92ea-4103-9ea9-536252324481"}
Value:
{
  "responseId": "52d96a0d-92ea-4103-9ea9-536252324481",
  "deleted": "true",
  "deletedAt": "2021-10-21T12:55:54.000+0000",
  "operationDate": "2021-05-21T12:55:54.000+0000"
}

Diana, I've reproduced the same issue that you're seeing and I'm trying to figure out if it's a bug or incorrect config. Let me get back to you on this.

Ok, thanks a lot 🙂

Thanks @Dunith Dhanushka

@Yupeng Fu ^^

hmm this is not expected. cc @Qiaochu Liu

gotcha, let me take a look

@Diana Arnos The reason being the columns are not added to the

partialUpsertStrategies

. Partial upsert only update values for the columns configured

+1 to that suggestion

makes sense, we probably can add a config for the default strategy

gotcha, i can take this. we can use overide strategy for columns that user don’t specific any “partialUpsertStrategies”. is this correct?

@Diana Arnos The reason being the columns are not added to the 

partialUpsertStrategies

. Partial upsert only update values for the columns configured

@Jackie but I understand a partial upsert would understand to only update the columns it is configured to and if the column is not present on the record it shouldn't set it to null and overwrite the original values. The original value that didn't receive an update should be kept as is

@Diana Arnos In order to get the values filled from the previous record (partial upsert semantic), currently the columns have to be configured in the

partialUpsertStrategies

from the

upsertConfig

E.g.

"upsertConfig": {
      "mode": "PARTIAL",
      "partialUpsertStrategies": {
        "formId": "OVERWRITE",
        "channelId": "OVERWRITE",
        "channelPlatform": "OVERWRITE",
        "companyId": "OVERWRITE",
        "submitted": "OVERWRITE",
        "deleted": "OVERWRITE",
        "createdAt": "OVERWRITE"
        "deletedAt": "OVERWRITE"
      },
      "hashFunction": "NONE"
    },

@Jackie thanks, that solved my problem. This is not explicit in the docs and a couple of weeks back my config was working, so I felt kinda lost.

@Diana Arnos yes, as @Jackie mentioned, you can update tableConfig to meet your requirement. we have a map to store the fields for partial upsert. For field in the map

(0) if field not in the mergeStrategies map, it will always use new value even it's null.
(1) if previous record null, return new record.
(2) if previous record not null. new record null. then use old value
(3) if previous record, not null, new record not null. then merge

in your scenario, for fields that get overwrite by null values, they are (0) instead of (2). @Jackie @Yupeng Fu do you think we can optimize the default behavior for “fields not in the strategies map” so that user don’t need to configure the “IgnoreIfNull” fields?

@Qiaochu Liu Yes, we can put a default strategy for fields not in the strategies map. Also, we should add

IGNORE

as a strategy where user can choose to ignore the old value

gotcha, thanks. i’ll create a diff today.

Yeah, the config solved my problem and now things are working nicely 🎉 Thank you all for all the help and explanation 😄 This will allow my team to finish setting up staging environment for Pinot and after a couple of tests we will finally go to Prod 😄