Hi Team, We have an analytics use case with a special requirement where we provide dynamic columns to the user which need not to be defined beforehand while deciding the schema and we provide querying capabilities on those fields as well. We have been exploring pinot, it fits well except for these dynamic fields. To solve this we first explored json type columns but the performance was not up to the mark, now we are looking into dynamically changing schema and adding column whenever we see a new dynamic field (which should not happen frequently) while processing the record and then putting that record into pinot. I have a few questions : 1. The number of records in the table when a new field appears can be a 100s of millions, would that be an issue when the schema changes or when segments are reloaded after schema change? 2. We are planning to keep sending records which do not have new fields to pinot even while we see some record with new field and we are processing schema changes for that. In pinot docs we found instructions to pause data consumption while changing schema. We are halting the records with new fields but if there is no new field in some other record we are continuing putting those records into pinot kafka topic. Can it result in corrupt data?

Did you use JSON index? It performs similar to inverted index, I am unclear why you didn’t get the performance

With this latest merged PR: I think the errors can also be avoided : https://github.com/apache/pinot/pull/15350

Yes, thanks

Hey guys. Thanks for the responses I work with Rishabh, and these questions came up as a result of my investigation into schema evolution. A little background may help here. The data we are recording has both fixed, and user-defined (custom) fields. We don't know the user-defined fields up front and new ones could be added for a variety of reasons. At first, he investigated using a JSON field, which allowed us a fixed schema, which was obviously simpler, but query response time was not what we were hoping for. So, he then tried "flattening out" the JSON, storing each key in JSON as a separate column, where he noticed a significant improvement in query performance. Perhaps that was because we did not define the appropriate indexes, not sure. Also, another feature of our data is that it is NOT immutable, including the keys in the JSON map. We may have to add new ones, remove old ones, or update existing ones. With JSON, we didn't see any way of doing that without having to load the current record, update the JSON, and write it back out. So, for us, at this point, there are really 2 main areas of unknowns: 1. With schema evolution, do we have to reload ALL segments when adding new columns and do we have to pause ALL data consumption, even for records that do not contain any of the new columns being added? 2. With Upsert support, can we update a record ANYTIME after first writing it, or is it only available for Upsert while it is in current in-memory segment? Also, I believe that it must also be recorded in the in-memory metadata, correct, so the retention here would define how long we could upsert a record? Should have mentioned that we are only using Realtime tables

@Gerald Bonfiglio I am questioning why JSON performance wasn't good. Did you have the right JSON indexing setup? If so, it should be the same as inverted index. The only tradeoff of JSON is that columnar will give you better compression for storage.

OK, thanks. Addresses my immediate concerns. As for JSON column, I wasn't involved with the performance evaluations, so not sure. But as I mentioned, JSON seems to have an issue with "incremental" updates, unless there is something in API that we have not discovered yet.

@Mayank, I used JSON index while using JSON column, without it the query did not return quickly, With Json index it took 46638 ms, The query involved 5-6 columns from json column referenced using JSON_EXTRACT_SCALAR, and it involved group by on multiple fields. While with flattening map into native dynamic columns with the help of schema evolution, it took less than 350 ms. Below is how we added Json indexes: "fieldConfigList": [ { "name": "PROPERTIES", "encodingType": "RAW", "indexes": { "json": { "indexPaths": [ "$.@level1", "$.@level5", "$.MeanRate", "$.type", "$.name" ], "maxLevels": 2, "excludeArray": false, "disableCrossArrayUnnest": true } } } ]

Can you share the query and the response metadata?

@Mayank, We are trying to do a bucketing on time in this query: SELECT CONCAT( DATETIMECONVERT( DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ), '1HOURSEPOCH', '1MILLISECONDSSIMPLE_DATE_FORMAT:yyyy-MM-dd HHmmss', '1:MILLISECONDS' ), DATETIMECONVERT( DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ) + 1, '1HOURSEPOCH', '1MILLISECONDSSIMPLE_DATE_FORMATHHmm:ss', '1:MILLISECONDS' ),' — ' ) AS hour_bucket, JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level1', 'STRING', 'N/A') AS KafkaCluster, JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level5', 'STRING', 'N/A') AS BrokerID, AVG(CAST(JSON_EXTRACT_SCALAR(PROPERTIES, '$.MeanRate', 'STRING', '0') AS DOUBLE)) AS BytesOutPerSec FROM snapshot WHERE JSON_EXTRACT_SCALAR(PROPERTIES, '$.type', 'STRING', 'N/A') = 'BrokerTopicMetrics' AND JSON_EXTRACT_SCALAR(PROPERTIES, '$.name', 'STRING', 'N/A') = 'BytesOutPerSec' AND JSON_EXTRACT_SCALAR(PROPERTIES, '$.MeanRate', 'STRING') IS NOT NULL GROUP BY JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level1', 'STRING', 'N/A'), JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level5', 'STRING', 'N/A'), DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ) ORDER BY KafkaCluster, BrokerID, DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ) Metadata: "numRowsResultSet": 10, "partialResult": false, "exceptions": [], "numGroupsLimitReached": false, "timeUsedMs": 42935, "requestId": "223284104000000324", "brokerId": "Broker_xxxxx", "numDocsScanned": 144342, "totalDocs": 13247643, "numEntriesScannedInFilter": 39626555, "numEntriesScannedPostFilter": 288684, "numServersQueried": 1, "numServersResponded": 1, "numSegmentsQueried": 16, "numSegmentsProcessed": 15, "numSegmentsMatched": 15, "numConsumingSegmentsQueried": 1, "numConsumingSegmentsProcessed": 0, "numConsumingSegmentsMatched": 0, "minConsumingFreshnessTimeMs": 1755605180789, "numSegmentsPrunedByBroker": 0, "numSegmentsPrunedByServer": 1, "numSegmentsPrunedInvalid": 0, "numSegmentsPrunedByLimit": 0, "numSegmentsPrunedByValue": 0, "brokerReduceTimeMs": 3, "offlineThreadCpuTimeNs": 0, "realtimeThreadCpuTimeNs": 0, "offlineSystemActivitiesCpuTimeNs": 0, "realtimeSystemActivitiesCpuTimeNs": 0, "offlineResponseSerializationCpuTimeNs": 0, "realtimeResponseSerializationCpuTimeNs": 0, "offlineTotalCpuTimeNs": 0, "realtimeTotalCpuTimeNs": 0, "explainPlanNumEmptyFilterSegments": 0, "explainPlanNumMatchAllFilterSegments": 0, "traceInfo": {}, "tablesQueried": [ "snapshot" ]

You need to use JSON_MATCH for predicates. And another reasons is Because of transform functions the indexing is not effective, check the entriesScannedInFilter is too high. It is not JSON, it is the transform functions.

@Mayank, below is the updated query with JSON_MATCH, SELECT CONCAT( DATETIMECONVERT( DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ), '1HOURSEPOCH', '1MILLISECONDSSIMPLE_DATE_FORMAT:yyyy-MM-dd HHmmss', '1:MILLISECONDS' ), DATETIMECONVERT( DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ) + 1, '1HOURSEPOCH', '1MILLISECONDSSIMPLE_DATE_FORMATHHmm:ss', '1:MILLISECONDS' ), ' — ' ) AS hour_bucket, JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level1', 'STRING', 'N/A') AS KafkaCluster, JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level5', 'STRING', 'N/A') AS BrokerID, AVG( CAST( JSON_EXTRACT_SCALAR(PROPERTIES, '$.MeanRate', 'STRING', '0') AS DOUBLE ) ) AS BytesOutPerSec FROM snapshot WHERE JSON_MATCH(PROPERTIES, '"$.type" = ''BrokerTopicMetrics''') AND JSON_MATCH(PROPERTIES, '"$.name" = ''BytesOutPerSec''') AND JSON_MATCH(PROPERTIES, '"$.MeanRate" IS NOT NULL') GROUP BY JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level1', 'STRING', 'N/A'), JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level5', 'STRING', 'N/A'), DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ) ORDER BY KafkaCluster, BrokerID, DATETIMECONVERT( SNAPSHOT_TIME, '1MILLISECONDSEPOCH', '1HOURSEPOCH', '1:HOURS' ); 1. This query return 0 records, while one with json extract scaler returns records.....I can see in segments, there is true against JSON Index against Property column 2. And by transform functions, Do you mean transform: CONCAT(DATETIMECONVERT(...), DATETIMECONVERT(...), ' — ') transform: JSON_EXTRACT_SCALAR(PROPERTIES, '$.@level1', 'STRING', 'N/A') transform: CAST(JSON_EXTRACT_SCALAR(PROPERTIES, '$.MeanRate', 'STRING', '0') AS DOUBLE) if so, I have same functions when using flattened map as native columns, How the query time is so less there then? Thanks!