appreciation post: Earlier today we launched our product www.sim.io (post), Pinot has been pretty central to us and will be to our users, thanks @Kishore G @Mayank @Jackie and the rest of the team for their support.

For those interested in search, we wrote a post on our learnings on full text search in Postgres vs Elastic: https://blog.paradedb.com/pages/elasticsearch_vs_postgres

Hi Everyone, I have a doubt. How to decide the: 1. The number of rows 2. The segment size 3. flush threshold time What should be the parameters for these judgements?

I have one doubt. Let say my consuming segment configuration is set to to certain values. 1. Size: 300 MB 2. Rows: 500k 3. Time: 24 Hours Then I want to change the rows and bring it down to 100k. So, now I made that change. So will my consuming segment which has consumed more than that number of rows will perform the commit? If yes, it will be after I perform rebalance of servers?

Hi community, I'm currently using Pinot 1.0.0 with a Kafka real-time table setup, and I have a use case where I would like to trigger an action immediately after a message is consumed by the Pinot real-time table. Is there a way for Pinot to acknowledge the consumption of a Kafka message, allowing me to execute a specific action upon successful ingestion of that message ? Any suggestions or best practices on how to achieve this would be greatly appreciated !

📣 Hello all! The Open Source Analytics Conference 2024 is next week! 🚀 Some really cool talks on databases, orchestration, Bi/visualization tools! • When: Nov 19-21 • Where: Online! • More info: osacon.io Hope to see you there!

Hi, does anyone know of ways to dynamically construct SQL queries for Pinot? Our data is spread across multiple tables, and we're looking for an efficient way to query it. Thanks!

Hey all! We use relational database as the source of truth and we are looking for options to replicate the data to Pinot as real-time as possible. We may need some transformations in-between. We don't have a "large" dataset, mostly under 1TB. What options do we have? As a side note we want to avoid CDC.

Hi everyone, I am new here, just want to explore features of pinot. I have use case where customer realtime transactions are coming in Kafka. and i do like to show last 1 minute/hours/seconds transaction total amount and number on a dashboard how can i achieve this with apache pinot.

Cheers Pinot🍷 Afficionados! I am trying to convince people in my org to consider Pinot for an analytics platform that can serve results directly to user app (web/ios/android). Description For an online marketplace, the use cases we are trying to achieve are 1. view counts of a post 2. count of connection requests that originated after viewing a particular post 3. list of 10 recently viewed posts. we intend to show counts of views each time a post is served. The source for this information is a stream of events that says which user viewed which post. Load Details We get 25000 such events every minute. that is also roughly the traffic we want to support that will request this data(queries) Questions • Is Pinot even a good candidate to solve this problem for just three use cases like above? (From what I have read so far, use-case 3 is not possible with sub-second performance without a key-value store to serve this data. But other 2 are achievable. ) • If someone has achieved this kind of use-case can you please correct any incorrect assumptions I am making? • What kind of costs can we estimate for a pinot cluster that will have this data? I am told that over a year we expect ingested stream of data to total <1TB. • I saw StarTree pricing and it seems that their premium tier supports 3tb of storage. So their managed service is sufficient for us? is this the correct lens to look at this problem? This is my first time trying to design a system like this and trying to get a buy-in for a system like this. All guidance and critique is highly appreciated, thanks for taking the time to read 🙂

So, 25,000 requests per minute ≈ 420 queries per second. Though it also depends on whether there is a 1:1 correlation for an "event" to a "query." It might be that for each "event" you need to actually fire off multiple queries. In any regard, that is precisely the kind of aggregations and filtering that Apache Pinot does super well at. LinkedIn does exactly that sort of thing with Pinot for post views and likes. And they have a billion+ users worldwide. I'm just spitballing here, but the 10 recently-viewed posts — might you be able to be solve this by using a Timestamp index, and then an ORDER BY and LIMIT? (SQL experts, please chime in if I am off-base.)

hi guys i am facing data ingestion issue after adding authentication to the pinot. does it have anything to do with the table config? i am getting error: Upload response: 500 Error: {"code":500,"error":"Caught exception when ingesting file into table: test_tenant_OFFLINE. Caught exception while uploading segments. Push mode: TAR, segment tars: [[file:/opt/pinot/tmp/ingestion_dir/working_dir_test_tenant_OFFLINE_1742312090194_FRKYyJJrdp/segment_tar_dir/test_tenant_1742312090217.tar.gz]]"} table_config = { "tableName": f"{tenant_name}", "tableType": "OFFLINE", "segmentsConfig": { "minimizeDataMovement": False, "deletedSegmentsRetentionPeriod": "7d", "retentionTimeUnit": "DAYS", "retentionTimeValue": "30", "segmentPushType": "REFRESH", "replication": "1" }, "tenants": { "broker": "DefaultTenant", "server": "DefaultTenant" }, "tableIndexConfig": { "invertedIndexColumns": [ "Time_of_Resolution", "Category", "Description", "Escalate", "Ticket_Status", "Customer_Gender", "Ticket_Subject", "product_purchased", "Source", "Relevance", "Customer_Email", "Customer_Name", "Tags", "Agent", "Number_of_conversations", "Intent", "Ticket_Priority", "Sentiment", "Type", "Created_date", "First_Response_Time", "Customer_Satisfaction_Rating", "Date_of_Purchase", "Resolution", "Created_date_month", "Created_date_year" ], "autoGeneratedInvertedIndex": False, "varLengthDictionaryColumns": [ "Time_of_Resolution", "Category", "Description", "Escalate", "Ticket_Status", "Customer_Gender", "Ticket_Subject", "product_purchased", "Source", "Relevance", "Customer_Email", "Customer_Name", "Tags", "Agent", "Intent", "Ticket_Priority", "Sentiment", "Type", "Created_date", "First_Response_Time", "Date_of_Purchase", "Resolution" ], "enableDefaultStarTree": False, "nullHandlingEnabled": True, "createInvertedIndexDuringSegmentGeneration": True, "rangeIndexVersion": 2, "aggregateMetrics": False, "optimizeDictionary": False, "loadMode": "MMAP", "enableDynamicStarTreeCreation": False, "columnMajorSegmentBuilderEnabled": False, "optimizeDictionaryForMetrics": False, "noDictionarySizeRatioThreshold": 0 }, "metadata": {}, "ingestionConfig": { "batchIngestionConfig": { "segmentIngestionType": "APPEND", "consistentDataPush": False }, "segmentTimeValueCheck": False, "transformConfigs": [ { "columnName": "First_Response_Time_epoch_millis", "transformFunction": "CASEWHEN(strcmp(Number_of_conversations,'null')=0,null,DATETIMECONVERT(First_Response_Time, 'SIMPLE_DATE_FORMAT|dd-MM-yyyy HH:mm', '1MILLISECONDSEPOCH', 'MINUTES|1'))" }, { "columnName": "Time_of_Resolution_epoch_millis", "transformFunction": "CASEWHEN(strcmp(Ticket_Status,'Closed')=0,DATETIMECONVERT(Time_of_Resolution, 'SIMPLE_DATE_FORMAT|dd-MM-yyyy HH:mm', '1MILLISECONDSEPOCH', 'MINUTES|1'),null)" }, { "columnName": "Created_date_timestamp", "transformFunction": "DATETIMECONVERT(Created_date, 'SIMPLE_DATE_FORMAT|dd-MM-yyyy HH:mm', 'SIMPLE_DATE_FORMAT|yyyy-MM-dd HH:mm', 'MINUTES|1')" }, { "columnName": "Created_date_epoch_millis", "transformFunction": "DATETIMECONVERT(Created_date, 'SIMPLE_DATE_FORMAT|dd-MM-yyyy HH:mm', '1MILLISECONDSEPOCH', 'MINUTES|1')" }, { "columnName": "Created_date_month", "transformFunction": "Month(Created_date_epoch_millis, 'UTC')" }, { "columnName": "Created_date_year", "transformFunction": "YEAR(Created_date_epoch_millis)" }, { "columnName": "Sentiment_score", "transformFunction": "CASEWHEN(strcmp(Sentiment,'Positive')=0,2,CASEWHEN(strcmp(Sentiment,'Neutral')=0,1,0))" }, { "columnName": "Sentiment_Positive", "transformFunction": "CASEWHEN(strcmp(Sentiment,'Positive')=0,2,0)" }, { "columnName": "Sentiment_Neutral", "transformFunction": "CASEWHEN(strcmp(Sentiment,'Neutral')=0,1,0)" }, { "columnName": "Sentiment_Negative", "transformFunction": "CASEWHEN(strcmp(Sentiment,'Negative')=0,0,0)" }, { "columnName": "Customer_Satisfaction_Rating", "transformFunction": "\"Customer Satisfaction Rating\"" }, { "columnName": "Number_of_conversations", "transformFunction": "\"Number of conversations\"" }, { "columnName": "Ticket_Subject", "transformFunction": "\"Ticket Subject\"" }, { "columnName": "Customer_Name", "transformFunction": "\"Customer Name\"" }, { "columnName": "Time_of_Resolution", "transformFunction": "\"Time to Resolution\"" }, { "columnName": "Customer_Age", "transformFunction": "\"Customer Age\"" }, { "columnName": "First_Response_Time", "transformFunction": "\"First Response Time\"" }, { "columnName": "Customer_Email", "transformFunction": "\"Customer Email\"" }, { "columnName": "Ticket_Status", "transformFunction": "\"Ticket Status\"" }, { "columnName": "Created_date", "transformFunction": "\"Created date\"" }, { "columnName": "Date_of_Purchase", "transformFunction": "\"Date of Purchase\"" }, { "columnName": "Agent_overall_performance", "transformFunction": "\"Agent overall performance\"" }, { "columnName": "Agent_politeness", "transformFunction": "\"Agent politeness\"" }, { "columnName": "Customer_Gender", "transformFunction": "\"Customer Gender\"" }, { "columnName": "Agent_communication", "transformFunction": "\"Agent communication\"" }, { "columnName": "Agent_Patience", "transformFunction": "\"Agent Patience\"" }, { "columnName": "Ticket_ID", "transformFunction": "\"Ticket ID\"" }, { "columnName": "Ticket_Priority", "transformFunction": "\"Ticket Priority\"" } ], "continueOnError": True, "rowTimeValueCheck": True }, "isDimTable": False }

please can anybody help i have been stuck for 4 days now!!!!!!!!!!alert the code i used was properly working fine with a namespace that didnt have authentication setup. when i did the authorization only i am facing this issue. the schema creation and table creation is working fine after authentication done in pinot. but the data ingestion is having issues. these are the config files: when i create the schema and table for a sample table. when i try ingest a sample csv file i cant do that: i am getting the above mentioned error in the chat. Upload response: 500 Error: {"code":500,"error":"Caught exception when ingesting file into table: test_tenant_OFFLINE. Caught exception while uploading segments. Push mode: TAR, segment tars: [[file:/opt/pinot/tmp/ingestion_dir/working_dir_test_tenant_OFFLINE_1742312090194_FRKYyJJrdp/segment_tar_dir/test_tenant_1742312090217.tar.gz]]"} ingest-pythoncode:

import requests
import json
from pathlib import Path
import time

PINOT_CONTROLLER = "********"
PINOT_BROKER = "*********"
AUTH_HEADERS = {
    "Authorization": "Basic YWRtaW46dmVyeXNlY3JldA==",
    "Content-Type": "application/json"
}

def verify_segment(tenant_name, segment_name):
    max_retries = 10
    retry_interval = 2  # seconds
    
    for i in range(max_retries):
        print(f"\nChecking segment status (attempt {i+1}/{max_retries})...")
        
        response = requests.get(f"{PINOT_CONTROLLER}/segments/{tenant_name}/{segment_name}/metadata",headers=AUTH_HEADERS)
        if response.status_code == 200:
            print("Segment is ready!")
            return True
            
        print(f"Segment not ready yet, waiting {retry_interval} seconds...")
        time.sleep(retry_interval)
    
    return False

def simple_ingest(tenant_name):
    # Verify schema and table exist
    schema_response = requests.get(f'{PINOT_CONTROLLER}/schemas/{tenant_name}',headers=AUTH_HEADERS)
    table_response = requests.get(f'{PINOT_CONTROLLER}/tables/{tenant_name}',headers=AUTH_HEADERS)
    
    if schema_response.status_code != 200 or table_response.status_code != 200:
        print(f"Schema or table missing for tenant {tenant_name}. Please run create_schema.py and create_table.py first")
        return

    csv_path = Path(f"data/{tenant_name}_data.csv")
    
    print(f"\nUploading data for tenant {tenant_name}...")
    with open(csv_path, 'rb') as f:
        files = {'file': (f'{tenant_name}_data.csv', f, 'text/csv')}
        
        # Using a dictionary for column mapping first
        column_map = {
            "Ticket ID": "Ticket_ID",
            "Customer Name": "Customer_Name",
            "Customer Email": "Customer_Email",
            "Company_name": "Company_name",
            "Customer Age": "Customer_Age",
            "Customer Gender": "Customer_Gender",
            "Product purchased": "product_purchased",
            "Date of Purchase": "Date_of_Purchase",
            "Ticket Subject": "Ticket_Subject",
            "Description": "Description",
            "Ticket Status": "Ticket_Status",
            "Resolution": "Resolution",
            "Ticket Priority": "Ticket_Priority",
            "Source": "Source",
            "Created date": "Created_date",
            "First Response Time": "First_Response_Time",
            "Time to Resolution": "Time_of_Resolution",
            "Number of conversations": "Number_of_conversations",
            "Customer Satisfaction Rating": "Customer_Satisfaction_Rating",
            "Category": "Category",
            "Intent": "Intent",
            "Type": "Type",
            "Relevance": "Relevance",
            "Escalate": "Escalate",
            "Sentiment": "Sentiment",
            "Tags": "Tags",
            "Agent": "Agent",
            "Agent politeness": "Agent_politeness",
            "Agent communication": "Agent_communication",
            "Agent Patience": "Agent_Patience",
            "Agent overall performance": "Agent_overall_performance",
            "Conversations": "Conversations"
        }

        config = {
            "inputFormat": "csv",
            "header": "true",
            "delimiter": ",",
            "fileFormat": "csv",
            "multiValueDelimiter": ";",
            "skipHeader": "false",
            # Convert dictionary to proper JSON string
            "columnHeaderMap": json.dumps(column_map)
        }
        
        params = {
            'tableNameWithType': f'{tenant_name}_OFFLINE',
            'batchConfigMapStr': json.dumps(config)
        }

        upload_headers = {
            "Authorization": AUTH_HEADERS["Authorization"]
        }
        
        response = <http://requests.post|requests.post>(
            f'{PINOT_CONTROLLER}/ingestFromFile',
            files=files,
            params=params,
            headers=upload_headers
        )
    
    print(f"Upload response: {response.status_code}")
    if response.status_code != 200:
        print(f"Error: {response.text}")
        return

    if response.status_code == 200:
        try:
            response_data = json.loads(response.text)
            print(f"Response data: {response_data}")
            segment_name = response_data["status"].split("segment: ")[1]
            print(f"\nWaiting for segment {segment_name} to be ready...")
            
            if verify_segment(tenant_name, segment_name):
                query = {
                    "sql": f"SELECT COUNT(*) FROM {tenant_name}_OFFLINE",
                    "trace": False
                }
                
                query_response = <http://requests.post|requests.post>(
                    f"{PINOT_BROKER}/query/sql",
                    json=query,
                    headers=AUTH_HEADERS
                )
                
                print("\nQuery response:", query_response.status_code)
                if query_response.status_code == 200:
                    print(json.dumps(query_response.json(), indent=2))
            else:
                print("Segment verification timed out")
        except Exception as e:
            print(f"Error processing segment: {e}")
            print(f"Full response text: {response.text}")

if __name__ == "__main__":
    simple_ingest("test_tenant")

Slack Conversation

Hello team, I am trying to set up S3 as the deep store for Pinot, but I’m facing issues. The configuration provided in the documentation is for version 0.6.0, and in this version, the joins are not working as expected. I want to use the latest version, but the configuration for that version isn’t working as it does in 0.6.0. Can you please assist me with this?

🎤 Got something exciting to share? The OSACon 2025 CFP is now officially open! 🚀 We're going online Nov 4–5, and we want YOU to be a part of it! Submit your proposal and be a speaker at the leading event for open-source analytics. 👉 Submit here: https://sessionize.com/osacon-2025/

This message was deleted.

Hello, everyone. I hope you are doing well. I am Blux Chang and looking for US-based developer. Feel free to contact me with DM.

New eBook Alert - "Building the Control Layer for Agentic AI with AI Gateway and MCP Servers" Learn how enterprises are deploying scalable, secure, and context-aware GenAI systems on-prem, in VPCs, or hybrid environments. A practical and technical roadmap for AI officers, Data scientists, Platform heads, MLOps engineers, CDOs, CIOs, and architects building production-grade AI infrastructure with AI Gateways and Model Context Protocol (MCP). Key takeaways are - Why AI Gateways are emerging as the central abstraction and orchestration layer in the GenAI stack How MCP allows agents to discover, authenticate, and invoke enterprise tools through machine-readable, context-aware schemas Deep dive into the architecture, protocols, and design principles shaping the next generation of enterprise AI infrastructure Access the free ungated version of the eBook here - https://gdurl.com/0RO8/download

For the interested, we wrote a blog post about the goods/bads of sync-ing Postgres via logical replication or via external ETL tools: https://www.paradedb.com/blog/etl_vs_logical_replication. Would love some feedback on it!

Hi everyone! I am organizing OSA Con 2025 and it's just around the corner. Lots of great speakers this year: Amazon, Snowflake, Preset, Percona, TiDB, Apple, Nutanix, Altinity, and more! 🗓️ Nov 4-5 📍ONLINE Register here: https://osacon.io/

🔥*Open Lakehouse & AI meetup in Chicago on Jan 22* Folks interested and/or using ClickHouse®, Apache Iceberg, lakehouse performance, or AI agents, this event is for you. Talks include: • Building a Foundation for AI with ClickHouse® + Iceberg — compute/storage separation, real perf results, roadmap, and how to get started (Robert Hodges, Altinity) • High-Performance Analytics on Iceberg — real bottlenecks and how modern engines fix them at scale (Chelsea Wang, CelerData) • Iceberg for Agents — turning lakehouse data into AI-ready context, structured RAG, and a live open-source demo (Andrew Madson, Fivetran) Expect real code and architecture and demos. 👉 https://luma.com/699ffm7v

Quick feedback and also a few retoric question : I’ve been working on improving the purgeTask to only be triguered on segments that match things with can be queried ! I’m definitly amazed by (with some help off course with the main entrypoints and the way to test) how claude code handle the task ! Got two plugins controller + minon with a docker-compose to try and debug the things. Get even more cool when using the MCP it grab a table and schema to my local test cluster to run it on real data ! Mind blowing But ... because there is a but question here : does it kills open source community(troubleshouting is getting quiet and slack slowly dying ?) ? Can this cool tools be used in a manner of improving community and maybe sharing things?

Hello Team We have an offline table and this table has all country data. We want to set different retention period each country. How can we achieve

👋 Hi team, I'm just trying to ideate around implementing a write-behind log for Pinot realtime ingestion. I wanted to see if there is any standard way of sending rows to WBL that has been ingested into real time table via Kafka to ensure downstreams that the row has been ingested.

Hey everyone, We are running Pinot 1.4 on Kubernetes. Currently, we store log files on EBS under the path

logs/pinotBroker

. Since this is ephemeral storage, the log files are lost whenever the pod restarts. We use custom Log4j configurations. Pinot is quite chatty, so we only send error logs to stdout in a structured format and ship them to Elasticsearch via Fluent Bit. Now we also want to persist: • component log files • query logs • audit logs Mainly so we can debug incidents later. Would it work if we simply changed

LOG_DIR

to point to an S3 path? • How would Pinot handle rolling log files in that case? • Would old files be deleted automatically? If S3 is not a good option, we can instead use a persistent EBS volume. • In that setup, does Pinot automatically clean up old log files periodically to avoid the volume filling up? Thanks!

Hi Team, A curious question !! From server 1 to 20. Call it as pinot001, 002 ..020.

If I create a segment, the primary is going to 001 and replica to 002.

Its always the same for all segments, if the segment is created in 013 its replica is in 014. I haven't done any instance assignment, this is the situation by default. Is this sequence for replica is voluntarily done, if yes, what is the reason behind?

Hey guys! We are evaluating https://docs.pinot.apache.org/operate-pinot/tuning/workload-query-isolation We essentially want to limit cpu and memory per user. We use Zk Basic Auth. There is no way currently to set these values on user level. We are thinking of creating one workload config per user. Our custom pinot proxy, will automatically inject

SET workloadName = USERNAME_workload;

at the top of every query. Is this the right approach and is it scalable? (let's say we have 1000 users) Thanks!

Hi Team, I see a setting like this in table definition under streamConfigMaps section.

realtime.segment.direct.realtime.offheap.allocation : true

Does this setting really exist ? If not why do pinot allow it to create the table in the first place?