https://pinot.apache.org/ logo
Join Slack
Powered by
Hello, I added a new column to my pinot schema and...
# general
s
Hello, I added a new column to my pinot schema and now i am. getting this -
Copy code
There are 8566 invalid segment/s. This usually means that they were created with an older schema. Please reload the table in order to refresh these segments to the new schema.
have tried reloading segments as well as rebalancing servers still getting the same warning messaghe.
m
Probably means reload failed. Can you check server logs?
s
here it is showing success
m
Interesting. @saurabh dubey if you can check it out?
s
@Sonit Rathi Was this column using a transform function? Pinot reports reload as successful if transform functions fail. Can you check the server logs for transformation failures?
s
There is a groovy transform function for the new added column but there are no errors in the logs.
s
Are these server logs? I see the pod says broker-1?
s
no the controller is running on this vm
s
We need server logs.. Not controller.
s
This is one of the servers. No errors in this as well
s
Have you added
Copy code
controller.disable.ingestion.groovy=false
To your controller conf? By default groovy is disabled
s
yes, as other older columns are also having transform function
s
I'd say, try reloading a single segment while tailing the server logs to look for clues.
s
Screenshot 2023-03-08 at 12.17.22 PM.png
@saurabh dubey any suggestions?
s
If a new column is indeed added, I'd expect to see
Copy code
Starting default column action: {} on column: {}
Log. Could you check the segment directory (default 
/home/pinot/data/index/{tableName}/{segmentName}
) And check if the new column that you've added is in the index_map file?
You'd want to look for something like
Copy code
<YourcolumnName>.forward_index.startOffset =
also metadata.properties for
Copy code
column.{yourColumnName}
s
the new column is not there in both
s
Can you share your schema and table config?
And the new column's name that you've added?
s
schema -
Copy code
{
  "schemaName": "bot_session_master_data",
  "dimensionFieldSpecs": [
    {
      "name": "brand_id",
      "dataType": "INT",
      "maxLength": 20
    },
    {
      "name": "bot_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "account_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "channel_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "customer_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "ext_session_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "flow_id",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "flow_version",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "status",
      "dataType": "INT",
      "maxLength": 10
    },
    {
      "name": "resolve_type",
      "dataType": "INT",
      "maxLength": 10
    },
    {
      "name": "last_node",
      "dataType": "STRING",
      "maxLength": 100
    },
    {
      "name": "ref_id",
      "dataType": "STRING",
      "maxLength": 100
    }
  ],
  "metricFieldSpecs": [
    {
      "name": "validity",
      "dataType": "INT",
      "maxLength": 100
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "created_at",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:SECONDS"
    }
  ]
}
table -
Copy code
{
  "REALTIME": {
    "tableName": "bot_session_master_data_REALTIME",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "schemaName": "bot_session_master_data",
      "retentionTimeUnit": "DAYS",
      "retentionTimeValue": "365",
      "replication": "2",
      "replicasPerPartition": "2",
      "minimizeDataMovement": false,
      "timeColumnName": "created_at"
    },
    "tenants": {
      "broker": "DefaultTenant",
      "server": "DefaultTenant",
      "tagOverrideConfig": {}
    },
    "tableIndexConfig": {
      "invertedIndexColumns": [
        "status",
        "channel_id",
        "resolve_type"
      ],
      "noDictionaryColumns": [],
      "rangeIndexVersion": 2,
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "sortedColumn": [
        "id",
        "customer_id"
      ],
      "rangeIndexColumns": [],
      "bloomFilterColumns": [],
      "loadMode": "MMAP",
      "streamConfigs": {
        "streamType": "kafka",
        "stream.kafka.topic.name": "bot-session-master-data",
        "stream.kafka.broker.list": "<http://infra-in-mc-kafkav2-01.onedirect.in:9092,infra-in-mc-kafkav2-02.onedirect.in:9092,infra-in-mc-kafkav2-03.onedirect.in:9092,infra-in-mc-kafkav2-04.onedirect.in:9092,infra-in-mc-kafkav2-05.onedirect.in:9092|infra-in-mc-kafkav2-01.onedirect.in:9092,infra-in-mc-kafkav2-02.onedirect.in:9092,infra-in-mc-kafkav2-03.onedirect.in:9092,infra-in-mc-kafkav2-04.onedirect.in:9092,infra-in-mc-kafkav2-05.onedirect.in: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": "1000000",
        "realtime.segment.flush.threshold.time": "24h",
        "realtime.segment.flush.segment.size": "100M"
      },
      "onHeapDictionaryColumns": [],
      "varLengthDictionaryColumns": [],
      "enableDefaultStarTree": false,
      "enableDynamicStarTreeCreation": true,
      "aggregateMetrics": false,
      "nullHandlingEnabled": false,
      "optimizeDictionaryForMetrics": false,
      "noDictionarySizeRatioThreshold": 0
    },
    "metadata": {},
    "quota": {},
    "routing": {
      "instanceSelectorType": "strictReplicaGroup"
    },
    "query": {},
    "ingestionConfig": {
      "transformConfigs": [
        {
          "columnName": "brand_id",
          "transformFunction": "Groovy({brandId},brandId)"
        },
        {
          "columnName": "ext_session_id",
          "transformFunction": "Groovy({extSessionId},extSessionId)"
        },
        {
          "columnName": "customer_id",
          "transformFunction": "Groovy({customerId},customerId)"
        },
        {
          "columnName": "bot_id",
          "transformFunction": "Groovy({botId},botId)"
        },
        {
          "columnName": "created_at",
          "transformFunction": "Groovy({createdAt},createdAt)"
        },
        {
          "columnName": "account_id",
          "transformFunction": "Groovy({accountId},accountId)"
        },
        {
          "columnName": "channel_id",
          "transformFunction": "Groovy({channelId},channelId)"
        },
        {
          "columnName": "flow_id",
          "transformFunction": "Groovy({flowId},flowId)"
        },
        {
          "columnName": "flow_version",
          "transformFunction": "Groovy({flowVersion},flowVersion)"
        },
        {
          "columnName": "resolve_type",
          "transformFunction": "Groovy({resolveType},resolveType)"
        },
        {
          "columnName": "last_node",
          "transformFunction": "Groovy({lastNode},lastNode)"
        },
        {
          "columnName": "ref_id",
          "transformFunction": "Groovy({refId},refId)"
        }
      ]
    },
    "isDimTable": false
  }
}
ref_id is the new column
s
Which pinot version are you on?
s
0.11
s
By any chance, the particular segment you're reloading isn't empty right? Can you run select count(*) from <table> where $segmentName = '{segmentName}' to verify?
s
no it isn't empty
but when I do select * it shows no records
s
So your reload transform function will fail, since
Copy code
"columnName": "ref_id",
          "transformFunction": "Groovy({refId},refId)"
references refId, which does not exist in the segment at all. i suppose the intent is that it'll be part of the incoming message? But at the very least I'd expect to see 2 log lines 1)
Copy code
<http://LOGGER.info|LOGGER.info>("Starting default column action: {} on column: {}", action, column);
2)
Copy code
LOGGER.warn("Skip creating derived column: {} because argument: {} does not exist in the segment", column,
                  argument);
I'm not sure why that's not being printed. Let me run some tests locally and will update
since existing segments won't have 
refId
to be able to evaluate the transform function, what is your expectation for the default value to be for old segments? I'm thinking you can follow these steps 1. Pause consumption (//tables/{tableName}/pauseConsumption) 2. Wait for pause consumption to finish (/tables/{tableName}/pauseStatus to return 0 consumingSegments) 3. Remove the transform function from your table config, trigger reload. This will put a default value for 
ref_id
("null") 4. Add the transform function back to table config, resume consumption. This would mean, ref_id will be set to null for all old segments, and after 4), new messages that will be ingested will have the correct value (parsed from message's refId field)
👍 1
If you want old segments to also have the correct value from the message, you have no other option than to delete all segments and re consume from scratch..
s
Heyy @saurabh dubey, this worked out great. No more warning msg and I am able to select the rows previous to schema change. Thanks a lot!!
s
Good to know @Sonit Rathi. Will discuss with the team and make changes to the reload status response to let users know when a transform is silently being dropped.
m
Can we add this in the FAQ until we have a better ux @saurabh dubey
s
Makes sense @Mayank. Let me go ahead and do that
👍 1
🙏 1
6 Views
Open in Slack
PreviousNext
Powered by