We run Pinot 0.8.0. When ingesting a table in

FULL

upsert

mode, we notice the number of rows returned for the same query varies across times, but it is supposed to remain consistent. For example, there are 1000 unique values keyed on column

A

, which we use as the primary key for the pinot table

table1

. A query like

select count(1) from table1

can return values 1567, or 789, in addition to 1000. In the case of 2000, you can find duplicated rows with different timestamps such as

| A | currenttime |
| - | ------------ |
| a | 1:00:00 |
| a | 1:00:01 |
| b | 1:00:00 |
| b | 1:00:03 |
...

In the case of 789, many rows are simply missing… We suspect this is related to the process of updating the index for the upserted table. Have anyone seen this before?

@Yupeng Fu ^^

can you check if the ingested topic is partitioned correctly?

the topic only has 1 partition and 1 replica per partition

if there are duplicates, that means the upsert is not configured right

@Yupeng Fu we only have 1 host for this testing in dev

and

$segmentName

to see which segments

@Yupeng Fu they are from different segments

it'll be helpful to see your table config and schema

@Yupeng Fu here is the minimal tableconfig:

{
  "tableName": "table1",
  "schema": {
    "metricFieldSpecs": [
      {
        "name": "B",
        "dataType": "DOUBLE"
      }
    ],
    "dimensionFieldSpecs": [
      {
        "name": "A",
        "dataType": "STRING"
      }
    ],
    "dateTimeFieldSpecs": [
      {
        "name": "EPOCH",
        "dataType": "INT",
        "format": "1:SECONDS:EPOCH",
        "granularity": "1:SECONDS"
      }
    ],
    "primaryKeyColumns": [
      "A"
    ],
    "schemaName": "schema1"
  },
  "realtime": {
    "tableName": "table1",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "schemaName": "schema1",
      "timeColumnName": "EPOCH",
      "replicasPerPartition": "1",
      "retentionTimeUnit": "DAYS",
      "retentionTimeValue": "4",
      "segmentPushType": "APPEND",
      "completionConfig": {
        "completionMode": "DOWNLOAD"
      }
    },
    "tableIndexConfig": {
      "invertedIndexColumns": [
        "A"
      ],
      "loadMode": "MMAP",
      "nullHandlingEnabled": false,
      "streamConfigs": {
        "realtime.segment.flush.threshold.rows": "10000",
        "realtime.segment.flush.threshold.time": "96h",
        "streamType": "kafka",
        "stream.kafka.consumer.type": "lowLevel",
        "stream.kafka.topic.name": "topic1",
        "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.broker.list": "kafka:9092",
        "stream.kafka.consumer.prop.auto.offset.reset": "largest"
      }
    },
    "tenants": {},
    "metadata": {},
    "routing": {
      "instanceSelectorType": "strictReplicaGroup"
    },
    "upsertConfig": {
      "mode": "FULL"
    }
  }
}

Do you get correct results if you do a "passthrough" query? i.e.

select * from pinot.default."select count(*) from <table>"

Message feed has stopped publishing so will have to test tomorrow. Just wanted to point it out that it’s not limited to

count()

- a regular

select * from table1

query without pushdowns can return more than and fewer than 1000 rows.

you can read the design here https://docs.google.com/document/d/1qljEMndPMxbbKtjlVn9mn2toz7Qrk0TGQsHLfI--7h8/edit

Based on the tests so far in Trino,

select count(*) from table1

or `select * from pinot.default."select count() from table1"`returns a num of rows less than or equal to 1000, whereas `select count() from table1 where from_unixtime(B) > current_timestamp - interval '15' minute` can give a number less than, equal to, or greater than 1000.

What is the version of trino you are using?

Trino version is 362

hmm, it shall not be, the indexing building during segment load and update is atomic

Telling from this pseudocode in the desgin doc,

remove

and

put/add

don’t happen within one transaction and queries may very well return different results depending on the timing?

def ingestionWithUpdate(PrimaryKeyIndex sealedSegmentIndex,
        PrimaryKeyIndex mutableSegmentIndex,
        ValidDocIndex validDocIndex):
 
 while msg := kafkaConsumer.poll() do:
     if msg.primaryKey in mutableSegmentIndex then: 
        if msg.time > mutableSegmentIndex.get(msg.primaryKey).time then:
          mutableSegmentIndex.put(msg.primaryKey, IndexTuple(segmentName,docId,msg.time))
          updateValidDocIndex(mutableSegmentIndex.get(msg.primaryKey))
     else if msg.primaryKey in sealedSegmentIndex then: 
        if msg.time > sealedSegmentIndex.get(msg.primaryKey).time then:
          sealedSegmentIndex.remove(msg.primaryKey)
          mutableSegmentIndex.put(msg.primaryKey, IndexTuple(segmentName,docId,msg.time))
          updateValidDocIndex(sealedSegmentIndex.get(msg.primaryKey))
     else:
        mutableSegmentIndex.put(msg.primaryKey, IndexTuple(segmentName,docId,msg.time))

     if time to seal:
        convert mutableSegmentIndex to add to sealedSegmentIndex
        update docId of the current segment in validDocIndex

def updateValidDocIndex(IndexTuple tuple):
   validDocIndex.get(tuple.segmentName).remove(tuple.docId)
   validDocIndex.get(active_segmentName).add(new_docId)

@Map The remove/add happens in 2 steps, where we first remove (invalidate) the old record, then add the new one. Also, when executing a query, the

validDocIndex

is read when the segment is executed, so small inconsistency during record update is expected.

@Jackie there is no notion or use of transactions?

No, pinot does not do transactions / global synchronization due to performance reasons

@Jackie IMHO, this is not about strong or weak consistency. Instead, it is about correctness. It’s okay to have eventual consistency, but in between, you can still return the old value for the record to be updated instead of nothing.

Pinot is an append only data store as of now, and upsert is handled as marking the old record invalid and add the new record. The challenge here is that there is no easy way to make these 2 updates atomic if they are applied to 2 different segments without locking all the segments, which could cause huge performance degradation. Either we add new record first or invalidate old record first, we will see duplicate key or missing key, and the current implementation is to invalidate the old record first

Thanks for the quickaround on the PR @Jackie! Very much appreciated. I think this should solve the 99x case within one segment.

Thanks @Map

Yeah, will merge it soon

Nice!

@Jackie it seems you push a image build every day to dockerhub based on the latest commit. Do you know when we can expect such an image including your patch?

@Xiang Fu ^^

You can check when the last image is pushed, and expected 24 hours time period. Usually 2pm PDT.

Good morning @Jackie, we tested your patch but we can still have this off-by-one row count intra-segment

@Map Good finding! If you only have one segment per partition, then this should be the case

we are running 0.10.0-SNAPSHOT-0290d74fe-20211201-jdk11 including your commit

@Map It is a little bit tricky here because for partial upsert, we need to first update the record value, then add the row. Let me see if we can break it into 2 steps

yeah it would make sense to do it in two steps, decoupling the index updates from row additions so that additions can happen first

Do you have one single segment per partition?

yes, one single segment per partition and there is only one partition

For

count(*)

query, it simply reads the

validDocIds

bitmap and calculates the cardinality

yes, 7844 is included

@Map

validDocIds

is a

ThreadSafeMutableRoaringBitmap

, which always take a snapshot when read, so there should be no race condition around it

@Jackie any ideas on what else might go wrong?

No idea. I feel the inner segment inconsistency should be totally eliminated with these 2 PRs

Did another test over the weekend. We can say for sure that there is no inconsistency when there is no message coming in. So it has to be something with the consuming or querying while consuming.

It sounds like the bug is caused by partial update. If so, is it possible to disable the partial update?

@Weixiang SunWhat makes you come to this conclusion if you don’t mind me asking? We are using the full upsert and partial upsert is disabled. @Jackie ‘s second patch last week separates them further in the code base

@Map Sorry, this was my impression which is clarified by you. We saw the duplicate record problem with pinot 0.8.0. The duplicate is persisting. Do you see the similar problem?

We see duplicates as described in case 3 at https://github.com/apache/pinot/issues/7849

Did you check and see if the duplicate is temporary or permanent? It seems permanent to me

temporary for us

thanks @Map

@Weixiang Sun did you figure out your problem? We are still experiencing the issue but also out of ideas on what could be going wrong… 😞

@Map we have wrong configuration. The duplicate records are gone after the configuration is fixed. It seems that we have the different problem.

Specifically we set

numPartitions

to 1 and

numInstancesPerPartition

to 0

Great! Thanks @Elon! But we are not eliminating the possibilities of the first two issues @Map pointed out.