New data keep saving in old format after changing schema

[RuslanTM]

Hey there!
I am researching schema evolution in pinot.
I use pinot, kafka and avro schema registry.
Messages in kafka are sending in avro format.

I have created a pinot realtime table that based on avro schema:

{"namespace": "some.domen", "type": "record", "name": "Product", "fields": [ {"type": "double", "name": "productID"}, {"name": "name", "type": "string"}, {"name": "published_date", "type": ["string", "null"]}, {"name": "cost", "type": "double"}, {"name": "timestamp", "type": "long"} ] }

First I converted avro schema into pinot schema and then created a pinot table using this config file:

{ "tableName": "product", "tableType": "REALTIME", "segmentsConfig": { "timeColumnName": "timestamp", "timeType": "MILLISECONDS", "schemaName": "product", "replicasPerPartition": "1" }, "tenants": {}, "tableIndexConfig": { "loadMode": "MMAP", "streamConfigs": { "streamType": "kafka", "stream.kafka.consumer.type": "lowlevel", "stream.kafka.topic.name": "product-topic", "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.avro.confluent.KafkaConfluentSchemaRegistryAvroMessageDecoder", "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory", "stream.kafka.zk.broker.url": "localhost:2181/kafka", "stream.kafka.broker.list": "localhost:9092", "realtime.segment.flush.threshold.time": "3600000", "realtime.segment.flush.threshold.size": "50000", "stream.kafka.consumer.prop.auto.offset.reset": "smallest", "stream.kafka.decoder.prop.schema.registry.rest.url": "http://localhost:8081" } }, "metadata": { "customConfigs": {} } }

After that I added a new column 'country' to avro schema. Converted into pinot schema too and updated the existing schema:

./bin/pinot-admin.sh UpdateSchema -schemaFile /tmp/schemas/product.json -exec

Then I executed:

curl -X POST "http://localhost:9000/segments/product/reload?type=REALTIME" -H "accept: application/json"

New column successfully added, I see null values in country column.
But new data that have not empty country column are importing with null values into pinot table too.

Is it a bug? Is it possible to import new data from kafka with new schema format without reloading pinot?

[xiangfu0]

Please this this doc: https://docs.pinot.apache.org/users/tutorials/schema-evolution#update-the-schema

Real-Time Pinot table: In case of real-time tables, make sure the
"pinot.server.instance.reload.consumingSegment" config is set to true inside Server config. 
Without this, the current consuming segment(s) will not reflect the default null value for newly added columns.

[RuslanTM]

@xiangfu0 I set this property and everything the same:

In kafka topic you can see Product_823 and Product_973 have country = USA:

But in pinot table country field is imported as null:

[Jackie-Jiang]

There are 2 ways of handling the schema change with new columns within consuming segment:

1. (Supported) Keep the already consumed records and add new columns as default values. The next consuming segment will pick up the new schema and consume the actual data. With this approach, the consumed records are kept, but we have to wait for the next consuming segment to ingest the actual data for the new columns (we might be able to build an api to force commit the current consuming segments and create new consuming segments)
2. (Not supported yet) Drop the already consumed records and re-consume from the start offset with the new columns with actual data. This approach can load data for new columns immediately, but the already consumed records are dropped and re-consumed which can cause temporary data inconsistency.

[RuslanTM]

@Jackie-Jiang the first approach is more preferable. Is it supported in the last version of pinot?

[Jackie-Jiang]

@RuslanTM Yes it is already supported. As mentioned above, you need to wait until the current consuming segment committed, then the next consuming segment will consume the actual data for the new added columns. The current consuming segment will be filled with default value for the new added columns, which is "null" for new column country.