diff --git a/frontend/src2/dashboard/Dashboard.vue b/frontend/src2/dashboard/Dashboard.vue
index 1efa428cf..e5eb76995 100644
--- a/frontend/src2/dashboard/Dashboard.vue
+++ b/frontend/src2/dashboard/Dashboard.vue
@@ -1,14 +1,14 @@
 <script setup lang="ts">
+import { useStorage } from '@vueuse/core'
 import { Breadcrumbs, call } from 'frappe-ui'
 import { RefreshCcw } from 'lucide-vue-next'
-import { computed, provide, ref } from 'vue'
+import { onBeforeUnmount, provide, ref } from 'vue'
 import { useRouter } from 'vue-router'
-import { downloadImage, waitUntil, wheneverChanges } from '../helpers'
-import useDashboard from './dashboard'
+import { downloadImage, waitUntil } from '../helpers'
 import { __ } from '../translation'
+import useDashboard from './dashboard'
 import DashboardItem from './DashboardItem.vue'
 import VueGridLayout from './VueGridLayout.vue'
-import { useStorage } from '@vueuse/core'
 
 const props = defineProps<{ name: string }>()
 
@@ -20,6 +20,11 @@ const dashboard = useDashboard(dashboard_name)
 provide('dashboard', dashboard)
 dashboard.refresh()
 
+// avoid listening if user navigates to another dashboard room
+onBeforeUnmount(() => {
+	dashboard.unsubscribeFromDashboardRoom()
+})
+
 const router = useRouter()
 function openWorkbook() {
 	router.push(`/workbook/${dashboard.doc.workbook}`)
diff --git a/frontend/src2/dashboard/dashboard.ts b/frontend/src2/dashboard/dashboard.ts
index 86827823c..ddfde6af7 100644
--- a/frontend/src2/dashboard/dashboard.ts
+++ b/frontend/src2/dashboard/dashboard.ts
@@ -11,6 +11,7 @@ import {
 	wheneverChanges,
 } from '../helpers'
 import useDocumentResource from '../helpers/resource'
+import { getSocket } from '../socket'
 import { isFilterValid } from '../query/components/filter_utils'
 import { column, filter_group } from '../query/helpers'
 import session from '../session'
@@ -51,16 +52,13 @@ function makeDashboard(name: string) {
 		return editing.value && editingItemIndex.value === dashboard.doc.items.indexOf(item)
 	}
 
-
 	const filters = ref<Record<string, FilterArgs[]>>({})
 	const filterStates = ref<Record<string, FilterState>>({})
 
 	function addChart(charts: WorkbookChart[]) {
 		const maxY = getMaxY()
 		charts.forEach((chart) => {
-			if (
-				!dashboard.doc.items.some((item) => item.type === 'chart' && item.chart === chart.name)
-			) {
+			if (!dashboard.doc.items.some((item) => item.type === 'chart' && item.chart === chart.name)) {
 				dashboard.doc.items.push({
 					type: 'chart',
 					chart: chart.name,
@@ -122,9 +120,7 @@ function makeDashboard(name: string) {
 
 	function positionNewFilter(newFilter: WorkbookDashboardItem) {
 		const items = dashboard.doc.items
-		const existingFilters = items.filter(
-			(item) => item.type === 'filter' && item !== newFilter
-		)
+		const existingFilters = items.filter((item) => item.type === 'filter' && item !== newFilter)
 
 		if (existingFilters.length === 0) {
 			newFilter.layout.x = 0
@@ -205,6 +201,22 @@ function makeDashboard(name: string) {
 		})
 	}
 
+	function subscribeToDashboardRoom() {
+		const socket = getSocket()
+
+		socket.emit('doc_subscribe', 'Insights Dashboard v3', name)
+		socket.on('connect', () => {
+			socket.emit('doc_subscribe', 'Insights Dashboard v3', name)
+		})
+	}
+
+	function unsubscribeFromDashboardRoom() {
+		const socket = getSocket()
+		socket.emit('doc_unsubscribe', 'Insights Dashboard v3', name)
+	}
+
+	subscribeToDashboardRoom()
+
 	function refresh(force = false) {
 		dashboard.doc.items
 			.filter((item) => item.type === 'chart')
@@ -214,6 +226,7 @@ function makeDashboard(name: string) {
 	function refreshChart(chart_name: string, force = false) {
 		const chart = useChart(chart_name)
 		chart.dataQuery.adhocFilters = getAdhocFilters(chart_name)
+		chart.dataQuery.dashboardName = name
 		chart.refresh(force)
 	}
 
@@ -336,6 +349,20 @@ function makeDashboard(name: string) {
 			.then(() => dashboard.load())
 	}
 
+	const defaultFilters = dashboard.doc.items.reduce((acc, item) => {
+		if (item.type != 'filter') return acc
+
+		const filterItem = item as WorkbookDashboardFilter
+		if (filterItem.default_operator && filterItem.default_value) {
+			acc[filterItem.filter_name] = {
+				operator: filterItem.default_operator,
+				value: filterItem.default_value,
+			}
+		}
+		return acc
+	}, {} as typeof filterStates.value)
+
+	Object.assign(filterStates.value, defaultFilters)
 
 	const key = `insights:dashboard-filter-states-${name}`
 	filterStates.value = store(key, () => filterStates.value)
@@ -399,6 +426,7 @@ function makeDashboard(name: string) {
 		updateAccess,
 
 		getShareLink,
+		unsubscribeFromDashboardRoom,
 	})
 }
 
@@ -431,13 +459,16 @@ function getDashboardResource(name: string) {
 		},
 	})
 	if (session.isLoggedIn) {
-		dashboard.onAfterLoad(() => dashboard.call('track_view').catch(() => { }))
+		dashboard.onAfterLoad(() => dashboard.call('track_view').catch(() => {}))
 	}
-	wheneverChanges(() => dashboard.doc.read_only, () => {
-		if (dashboard.doc.read_only) {
-			dashboard.autoSave = false
+	wheneverChanges(
+		() => dashboard.doc.read_only,
+		() => {
+			if (dashboard.doc.read_only) {
+				dashboard.autoSave = false
+			}
 		}
-	})
+	)
 	return dashboard
 }
 
diff --git a/frontend/src2/query/query.ts b/frontend/src2/query/query.ts
index 9c6ff91b1..ce2217e6e 100644
--- a/frontend/src2/query/query.ts
+++ b/frontend/src2/query/query.ts
@@ -16,11 +16,13 @@ import { confirmDialog } from '../helpers/confirm_dialog'
 import { FIELDTYPES } from '../helpers/constants'
 import useDocumentResource from '../helpers/resource'
 import { createToast } from '../helpers/toasts'
-import { __ } from '../translation'
 import router from '../router'
 import session from '../session'
+import { getSocket } from '../socket'
+import { __ } from '../translation'
 import {
 	AdhocFilters,
+	aggregations,
 	CodeArgs,
 	ColumnDataType,
 	ColumnOption,
@@ -47,7 +49,6 @@ import {
 	Summarize,
 	SummarizeArgs,
 	UnionArgs,
-	aggregations,
 } from '../types/query.types'
 import { InsightsQueryv3, QueryVariable } from '../types/workbook.types'
 import useWorkbook from '../workbook/workbook'
@@ -155,7 +156,16 @@ export function makeQuery(name: string) {
 	}
 
 	const adhocFilters = ref<AdhocFilters>()
-	async function execute(force: boolean = false) {
+	// set before execute
+	// to get realtime completion events to the correct dashboard doc room
+	const dashboardName = ref<string>()
+	const pendingExecutionKey = ref<string | null>(null)
+
+	const REALTIME_TIMEOUT = 120_000
+	const QUEUE_RETRY_DELAY = 3000
+	const MAX_QUEUE_RETRIES = 30
+
+	async function execute(force: boolean = false, retryCount: number = 0): Promise<any> {
 		if (!query.islocal) {
 			await waitUntil(() => query.isloaded)
 		}
@@ -167,6 +177,7 @@ export function makeQuery(name: string) {
 
 		if (
 			!force &&
+			retryCount === 0 &&
 			lastExecutionArgs &&
 			isEqual(lastExecutionArgs, {
 				operations: currentOperations.value,
@@ -177,51 +188,126 @@ export function makeQuery(name: string) {
 		}
 
 		executing.value = true
+		pendingExecutionKey.value = null
+
 		return query
 			.call('execute', {
 				active_operation_idx: activeOperationIdx.value,
 				adhoc_filters: adhocFilters.value,
 				force,
+				dashboard: dashboardName.value,
 			})
-		.then((response: any) => {
-			if (!response) return
-
-			result.value.executedSQL = response.sql
-			result.value.columns = response.columns
-			result.value.rows = response.rows
-			result.value.totalRowCount = 0
-			result.value.formattedRows = getFormattedRows(result.value, query.doc.operations)
+			.then(async (response: any) => {
+				if (!response) return
+
+				if (response.status === 'pending') {
+					pendingExecutionKey.value = response.key
+					populateResults(response)
+					const completed = await waitForSocketEvent(response.key)
+					if (completed) {
+						pendingExecutionKey.value = null
+						return execute(false, retryCount)
+					}
+					return
+				}
 
-			const aggregationPrefixes = aggregations.map((a) => `${a}_`)
-			const isMeasureColumn = (name: string) =>
-				measureColumns.value.includes(name) ||
-				aggregationPrefixes.some((prefix) => name.startsWith(prefix))
+				if (response.status === 'queue_full') {
+					if (retryCount >= MAX_QUEUE_RETRIES) {
+						executing.value = false
+						createToast({
+							title: 'Query Queue Full',
+							message: 'Too many queries running, Please try again later',
+							variant: 'info',
+						})
+						return
+					}
 
-			result.value.columnOptions = result.value.columns.map((column) => {
-				return {
-					label: column.name,
-					value: column.name,
-					description: column.type,
-					query: query.doc.name,
-					data_type: column.type,
-					is_measure: isMeasureColumn(column.name),
+					await new Promise((resolve) => setTimeout(resolve, QUEUE_RETRY_DELAY))
+					return execute(force, retryCount + 1)
 				}
+
+				populateResults(response)
 			})
-			result.value.timeTaken = response.time_taken
-			result.value.lastExecutedAt = new Date()
-		})
 			.catch(() => {
 				result.value = { ...EMPTY_RESULT }
 			})
 			.finally(() => {
-				executing.value = false
-				lastExecutionArgs = {
-					operations: currentOperations.value,
-					adhoc_filters: adhocFilters.value,
+				if (!pendingExecutionKey.value) {
+					executing.value = false
+				}
+				if (retryCount === 0) {
+					lastExecutionArgs = {
+						operations: currentOperations.value,
+						adhoc_filters: adhocFilters.value,
+					}
 				}
 			})
 	}
 
+	function populateResults(response: any) {
+		result.value.executedSQL = response.sql
+		result.value.columns = response.columns
+
+		const aggregationPrefixes = aggregations.map((a) => `${a}_`)
+		const isMeasureColumn = (name: string) =>
+			measureColumns.value.includes(name) ||
+			aggregationPrefixes.some((prefix) => name.startsWith(prefix))
+
+		result.value.columnOptions = result.value.columns.map((column) => {
+			return {
+				label: column.name,
+				value: column.name,
+				description: column.type,
+				query: query.doc.name,
+				data_type: column.type,
+				is_measure: isMeasureColumn(column.name),
+			}
+		})
+
+		if (response.data?.length >= 0) {
+			result.value.timeTaken = response.time_taken
+			result.value.rows = response.data
+			result.value.totalRowCount = 0
+			result.value.formattedRows = getFormattedRows(result.value, query.doc.operations)
+		}
+		result.value.lastExecutedAt = new Date()
+	}
+
+	// waits for `insights_query_complete` socket event
+	// returns true if the event fired, false if timed out
+	function waitForSocketEvent(key: string): Promise<boolean> {
+		return new Promise<boolean>((resolve) => {
+			const sock = getSocket()
+
+			const timeout = setTimeout(() => {
+				cleanup()
+				if (pendingExecutionKey.value === key) {
+					pendingExecutionKey.value = null
+					executing.value = false
+					createToast({
+						title: 'Query Timeout',
+						message: 'The query is taking too long. Please try again later',
+						variant: 'warning',
+					})
+				}
+				resolve(false)
+			}, REALTIME_TIMEOUT)
+
+			function handler(data: any) {
+				if (data?.key !== key || pendingExecutionKey.value !== key) return
+				cleanup()
+				resolve(true)
+			}
+
+			function cleanup() {
+				sock.off('insights_query_complete', handler)
+				clearTimeout(timeout)
+			}
+
+			sock.on('insights_query_complete', handler)
+		})
+	}
+
 	const fetchingCount = ref(false)
 	async function fetchResultCount() {
 		if (!query.islocal) {
@@ -380,12 +466,12 @@ export function makeQuery(name: string) {
 			(op) =>
 				op.type === 'order_by' &&
 				op.column.column_name === args.column.column_name &&
-				op.direction === args.direction
+				op.direction === args.direction,
 		)
 		if (existingOrderBy) return
 
 		const existingOrderByIndex = currentOperations.value.findIndex(
-			(op) => op.type === 'order_by' && op.column.column_name === args.column.column_name
+			(op) => op.type === 'order_by' && op.column.column_name === args.column.column_name,
 		)
 		if (existingOrderByIndex > -1) {
 			query.doc.operations[existingOrderByIndex] = order_by(args)
@@ -396,7 +482,7 @@ export function makeQuery(name: string) {
 
 	function removeOrderBy(column_name: string) {
 		const index = query.doc.operations.findIndex(
-			(op) => op.type === 'order_by' && op.column.column_name === column_name
+			(op) => op.type === 'order_by' && op.column.column_name === column_name,
 		)
 		if (index > -1) {
 			query.doc.operations.splice(index, 1)
@@ -432,7 +518,7 @@ export function makeQuery(name: string) {
 	function renameColumn(oldName: string, newName: string) {
 		// Check if there's a mutate operation with the old name
 		const existingMutateIdx = currentOperations.value.findIndex(
-			(op) => op.type === 'mutate' && op.new_name === oldName
+			(op) => op.type === 'mutate' && op.new_name === oldName,
 		)
 
 		if (existingMutateIdx !== -1) {
@@ -446,7 +532,7 @@ export function makeQuery(name: string) {
 		}
 
 		const existingRenameIdx = currentOperations.value.findIndex(
-			(op) => op.type === 'rename' && op.new_name === oldName
+			(op) => op.type === 'rename' && op.new_name === oldName,
 		)
 
 		if (existingRenameIdx === -1) {
@@ -455,7 +541,7 @@ export function makeQuery(name: string) {
 				rename({
 					column: column(oldName),
 					new_name: newName,
-				})
+				}),
 			)
 			return
 		}
@@ -475,7 +561,7 @@ export function makeQuery(name: string) {
 			rename({
 				column: column(originalColumnName),
 				new_name: newName,
-			})
+			}),
 		)
 	}
 
@@ -498,7 +584,7 @@ export function makeQuery(name: string) {
 	function changeColumnType(column_name: string, newType: ColumnDataType) {
 		// Check if there's a mutate operation with the old name
 		const existingMutateIdx = currentOperations.value.findIndex(
-			(op) => op.type === 'mutate' && op.new_name === column_name
+			(op) => op.type === 'mutate' && op.new_name === column_name,
 		)
 
 		if (existingMutateIdx !== -1) {
@@ -515,7 +601,7 @@ export function makeQuery(name: string) {
 			cast({
 				column: column(column_name),
 				data_type: newType,
-			})
+			}),
 		)
 	}
 
@@ -820,10 +906,7 @@ export function makeQuery(name: string) {
 
 		// No local summarize/pivot — check if the source is another query
 		const sourceOp = operations.find((op) => op.type === 'source') as Source | undefined
-		if (
-			!sourceOp ||
-			sourceOp.table.type !== 'query'
-		) {
+		if (!sourceOp || sourceOp.table.type !== 'query') {
 			createToast({
 				title: __('Failed to drill down'),
 				message: __('Drill down is only supported on summarized data'),
@@ -852,18 +935,13 @@ export function makeQuery(name: string) {
 
 		// Merge: source query's operations + any filter_groups from the current query
 		// (filters applied on top of the source query should still be respected)
-		const mergedOps = [
-			...sourceOps,
-			...operations.filter((op) => op.type === 'filter_group'),
-		]
+		const mergedOps = [...sourceOps, ...operations.filter((op) => op.type === 'filter_group')]
 
 		// Recursively resolve — the source query might itself have a query source
-		return getEffectiveOperationsForDrillDown(
-			mergedOps,
-			currRow,
-			col,
-			[..._visitedQueryNames, sourceQueryName],
-		)
+		return getEffectiveOperationsForDrillDown(mergedOps, currRow, col, [
+			..._visitedQueryNames,
+			sourceQueryName,
+		])
 	}
 
 	function getFiltersForDimension(dim: Dimension, value: string) {
@@ -948,7 +1026,7 @@ export function makeQuery(name: string) {
 		operations: Operation[],
 		summarizeIdx: number,
 		col: QueryResultColumn,
-		row: QueryResultRow
+		row: QueryResultRow,
 	) {
 		const filters: FilterArgs[] = []
 		const summarizeOperation = operations[summarizeIdx] as Summarize
@@ -967,7 +1045,7 @@ export function makeQuery(name: string) {
 		operations: Operation[],
 		pivotIdx: number,
 		col: QueryResultColumn,
-		row: QueryResultRow
+		row: QueryResultRow,
 	) {
 		const pivotOperation = operations[pivotIdx] as PivotWiderArgs
 
@@ -999,7 +1077,6 @@ export function makeQuery(name: string) {
 		return filters
 	}
 
-
 	function copyQuery() {
 		query.call('export').then((data) => {
 			copyToClipboard(JSON.stringify(data, null, 2))
@@ -1040,7 +1117,7 @@ export function makeQuery(name: string) {
 			deep: true,
 			capacity: 100,
 			debounce: 500,
-		}
+		},
 	)
 
 	const importingTables = ref(false)
@@ -1084,7 +1161,7 @@ export function makeQuery(name: string) {
 					}
 				}
 			},
-			{ debounce: 500 }
+			{ debounce: 500 },
 		)
 	})
 
@@ -1097,6 +1174,7 @@ export function makeQuery(name: string) {
 		currentOperations,
 		activeEditOperation,
 		adhocFilters,
+		dashboardName,
 
 		autoExecute,
 		executing,
@@ -1199,7 +1277,7 @@ function getQueryResource(name: string) {
 			if (query.doc.read_only) {
 				query.autoSave = false
 			}
-		}
+		},
 	)
 	return query
 }
diff --git a/frontend/src2/settings/GeneralSettings.vue b/frontend/src2/settings/GeneralSettings.vue
index 27b8fa4b6..f3db3b5e1 100644
--- a/frontend/src2/settings/GeneralSettings.vue
+++ b/frontend/src2/settings/GeneralSettings.vue
@@ -1,10 +1,10 @@
 <script setup lang="ts">
 import { call } from 'frappe-ui'
-import { __ } from '../translation'
 import { ref } from 'vue'
 import { createToast } from '../helpers/toasts'
 import DatePickerControl from '../query/components/DatePickerControl.vue'
 import session from '../session'
+import { __ } from '../translation'
 import SettingItem from './SettingItem.vue'
 import useSettings from './settings'
 
@@ -80,6 +80,19 @@ async function setupDemoData() {
 			/>
 		</SettingItem>
 
+		<SettingItem
+			label="Max Concurrent Queries"
+			description="Maximum number of queries that can run at the same time. Default is 6 Maximum is 18"
+		>
+			<FormControl
+				class="w-28"
+				type="number"
+				v-model.number="settings.doc.max_concurrent_queries"
+				:min="6"
+				:max="18"
+			/>
+		</SettingItem>
+
 		<SettingItem
 			v-if="session.user.is_admin && !session.user.has_demo_data"
 			label="Demo Data"
diff --git a/frontend/src2/settings/settings.ts b/frontend/src2/settings/settings.ts
index 7a000f065..c1ff712ea 100644
--- a/frontend/src2/settings/settings.ts
+++ b/frontend/src2/settings/settings.ts
@@ -19,6 +19,7 @@ function makeSettings() {
 			allowed_origins: '',
 			max_records_to_sync: 10_00_000,
 			max_memory_usage: 512,
+			max_concurrent_queries: 6,
 			fiscal_year_start: '2024-04-01',
 			week_starts_on: 'Monday',
 			enable_data_store: false,
@@ -47,6 +48,7 @@ type InsightsSettings = {
 	allowed_origins: string
 	max_records_to_sync: number
 	max_memory_usage: number
+	max_concurrent_queries: number
 	fiscal_year_start: string
 	week_starts_on: string
 	enable_data_store: boolean
diff --git a/frontend/src2/socket.ts b/frontend/src2/socket.ts
index 5ba597ba6..70d32777d 100644
--- a/frontend/src2/socket.ts
+++ b/frontend/src2/socket.ts
@@ -2,15 +2,22 @@ import { io, Socket } from 'socket.io-client'
 
 let socket: Socket
 
+declare global {
+	interface Window {
+		site_name: string
+		socketio_port: number
+	}
+}
+
 export function getSocket() {
 	if (socket) {
 		return socket
 	}
 
-	let socketio_port = window.socketio_port || 9000
+	let socket_port = window.socketio_port || 9000
 	let host = window.location.hostname
 	let siteName = import.meta.env.DEV ? host : window.site_name
-	let port = window.location.port ? `:${socketio_port}` : ''
+	let port = window.location.port ? `:${socket_port}` : ''
 	let protocol = port ? 'http' : 'https'
 	let url = `${protocol}://${host}${port}/${siteName}`
 
diff --git a/insights/api/data_sources.py b/insights/api/data_sources.py
index 878ce5868..a27741eec 100644
--- a/insights/api/data_sources.py
+++ b/insights/api/data_sources.py
@@ -331,7 +331,7 @@ def get_data_source_table(data_source: str, table_name: str):
     check_table_permission(data_source, table_name)
     ds = frappe.get_doc("Insights Data Source v3", data_source)
     q = ds.get_ibis_table(table_name).head(100)
-    data, time_taken = execute_ibis_query(q, cache_expiry=24 * 60 * 60)
+    data, time_taken = execute_ibis_query(q, cache_expiry=24 * 60 * 60, use_lock=False)
 
     return {
         "table_name": table_name,
diff --git a/insights/insights/doctype/insights_data_source_v3/ibis_utils.py b/insights/insights/doctype/insights_data_source_v3/ibis_utils.py
index d1846a4f8..4be455f9a 100644
--- a/insights/insights/doctype/insights_data_source_v3/ibis_utils.py
+++ b/insights/insights/doctype/insights_data_source_v3/ibis_utils.py
@@ -3,6 +3,7 @@
 import time
 from contextlib import contextmanager
 from datetime import date
+from typing import Literal
 
 import frappe
 import ibis
@@ -31,6 +32,92 @@
 from .ibis.functions import fiscal_year_start, quarter_start, week_start
 from .ibis.utils import get_functions
 
+# RedLock constants
+QUERY_LOCK_PREFIX = "insights_query_lock:"
+SEMAPHORE_KEY = "insights_query_semaphore"
+DEFAULT_MAX_CONCURRENT_QUERIES = 6
+LOCK_TIMEOUT = 180
+
+
+def get_max_concurrent_queries():
+    value = (
+        frappe.db.get_single_value("Insights Settings", "max_concurrent_queries")
+        or DEFAULT_MAX_CONCURRENT_QUERIES
+    )
+    return max(2, min(18, int(value)))
+
+
+def try_acquire_lock(key) -> bool:
+    try:
+        key = frappe.cache.make_key(f"{QUERY_LOCK_PREFIX}{key}")
+        result = bool(frappe.cache.set(key, "1", ex=LOCK_TIMEOUT, nx=True))
+        return result
+    except Exception:
+        return False
+
+
+def release_lock(key):
+    try:
+        key = frappe.cache.make_key(f"{QUERY_LOCK_PREFIX}{key}")
+        frappe.cache.delete(key)
+    except Exception:
+        pass
+
+
+def try_acquire_semaphore():
+    try:
+        cache = frappe.cache()
+        key = cache.make_key(SEMAPHORE_KEY)
+        current = cache.incr(key)
+        cache.expire(key, LOCK_TIMEOUT)
+
+        if current > get_max_concurrent_queries():
+            cache.decr(key)
+            return None
+
+        return current
+    except Exception:
+        return 0
+
+
+def release_semaphore():
+    try:
+        cache = frappe.cache()
+        key = cache.make_key(SEMAPHORE_KEY)
+        current = cache.decr(key)
+
+        if current < 0:
+            cache.set(key, 0)
+
+    except Exception:
+        pass
+
+
+def is_query_executing(key):
+    try:
+        key = frappe.cache.make_key(f"{QUERY_LOCK_PREFIX}{key}")
+        t = frappe.cache.get(key) is not None
+        return t
+    except Exception:
+        return False
+
+
+# for frontend
+def get_pending_query_result(cache_key):
+    # fix: check if query is still executing first
+    # fixes issue where we check cache before its written
+    if is_query_executing(cache_key):
+        return {"status": "pending"}
+
+    if has_cached_results(cache_key):
+        result = get_cached_results(cache_key)
+        return {
+            "status": "completed",
+            "result": result,
+        }
+
+    return {"status": "not_found"}
+
 
 class IbisQueryBuilder:
     def __init__(self, doc, active_operation_idx=None):
@@ -741,6 +828,22 @@ def get_current_columns(self):
         return {col: getattr(self.query, col) for col in self.query.schema().names}
 
 
+def generate_query_key(query: IbisQuery):
+    sql = ibis.to_sql(query)
+    backends, _ = query._find_backends()
+    backend_id = backends[0].db_identity if backends else None
+    return make_digest(sql, backend_id), sql
+
+
+class QueryExecutionResponse(frappe._dict):
+    key: str
+    status: Literal["pending", "completed", "error", "queue_full"]
+    sql: str
+    columns: list[dict]
+    data: pd.DataFrame | None
+    time_taken: float
+
+
 def execute_ibis_query(
     query: IbisQuery,
     limit=100,
@@ -749,52 +852,90 @@ def execute_ibis_query(
     cache_expiry=3600,
     reference_doctype=None,
     reference_name=None,
-):
-    try:
-        sql = ibis.to_sql(query)
-    except ibis.common.exceptions.OperationNotDefinedError:
-        # TODO: throw better error message
-        raise
-
-    if cache:
-        backends, _ = query._find_backends()
-        backend_id = backends[0].db_identity if backends else None
-        cache_key = make_digest(sql, backend_id)
-
-        if has_cached_results(cache_key) and not force:
-            return get_cached_results(cache_key), -1
-
+    use_lock=True,
+) -> QueryExecutionResponse:
     if hasattr(query, "limit") and limit:
         limit = int(limit or 100)
         limit = min(max(limit, 1), 10_00_000)
         query = query.limit(limit)
 
+    key, sql = generate_query_key(query)
+    columns = get_columns_from_schema(query.schema())
+
+    if cache and has_cached_results(key) and not force:
+        df = get_cached_results(key)
+        return QueryExecutionResponse(
+            key=key,
+            status="completed",
+            data=df,
+            sql=sql,
+            time_taken=0,
+            columns=columns,
+        )
+
+    if not try_acquire_lock(key):
+        return QueryExecutionResponse(
+            key=key,
+            status="pending",
+            # sql=sql,
+            # columns=columns,
+            data=None,
+            time_taken=0,
+        )
+
+    slot = try_acquire_semaphore()
+    if slot is None:
+        # too many concurrent queries
+        release_lock(key)
+        return QueryExecutionResponse(
+            key=key,
+            status="queue_full",
+            # sql=sql,
+            # columns=columns,
+            data=None,
+            time_taken=0,
+        )
+
+    try:
+        df, time_taken = _execute_query(query, sql, reference_doctype, reference_name)
+
+        if cache:
+            cache_results(key, df, cache_expiry)
+
+        return QueryExecutionResponse(
+            key=key,
+            status="completed",
+            data=df,
+            sql=sql,
+            time_taken=time_taken,
+            columns=columns,
+        )
+    finally:
+        release_lock(key)
+        release_semaphore()
+
+
+def _execute_query(query: IbisQuery, sql, reference_doctype, reference_name) -> tuple[pd.DataFrame, float]:
     start = time.monotonic()
 
     try:
-        result = query.execute()
+        df = query.execute()
+        if isinstance(df, pd.DataFrame):
+            df = df.replace({pd.NaT: None, np.nan: None})
     except Exception as e:
         if "max_statement_time" in str(e):
-            frappe.log_error(
-                title="Query execution time exceeded the limit.",
-                message=f"Query: {sql}",
-            )
+            frappe.log_error("Query execution time exceeded the limit.")
             max_time = frappe.db.get_single_value("Insights Settings", "max_execution_time") or 180
             frappe.throw(
                 title="Query Timeout",
-                msg=f"Query execution time exceeded the limit of {max_time} seconds. Please try again with a smaller timespan or a more specific filter.",
+                msg=f"Query execution time exceeded the limit of {max_time} seconds. Please try again with a more specific filter.",
             )
         raise e
 
     time_taken = flt(time.monotonic() - start, 3)
     create_execution_log(sql, time_taken, reference_name)
 
-    if isinstance(result, pd.DataFrame):
-        result = result.replace({pd.NaT: None, np.nan: None})
-        if cache:
-            cache_results(cache_key, result, cache_expiry)
-
-    return result, time_taken
+    return df, time_taken
 
 
 def get_columns_from_schema(schema: ibis.Schema):
diff --git a/insights/insights/doctype/insights_query_v3/insights_query_v3.json b/insights/insights/doctype/insights_query_v3/insights_query_v3.json
index e69ce01b2..87711020e 100644
--- a/insights/insights/doctype/insights_query_v3/insights_query_v3.json
+++ b/insights/insights/doctype/insights_query_v3/insights_query_v3.json
@@ -155,4 +155,4 @@
  "states": [],
  "title_field": "title",
  "track_changes": 1
-}
\ No newline at end of file
+}
diff --git a/insights/insights/doctype/insights_query_v3/insights_query_v3.py b/insights/insights/doctype/insights_query_v3/insights_query_v3.py
index 0afab4bc6..5d046f09c 100644
--- a/insights/insights/doctype/insights_query_v3/insights_query_v3.py
+++ b/insights/insights/doctype/insights_query_v3/insights_query_v3.py
@@ -6,7 +6,6 @@
 from io import BytesIO
 
 import frappe
-import ibis
 import sqlparse
 from frappe.model.document import Document
 from ibis import _
@@ -70,6 +69,9 @@ def on_trash(self):
     def before_save(self):
         self.set_linked_queries()
 
+    def _get_execution_lock_key(self):
+        return f"insights_query_exec:{self.name}"
+
     def cleanup_empty_folder(self, folder_name):
         """Delete folder if it has no queries or charts"""
         if not frappe.db.exists("Insights Folder", folder_name):
@@ -161,34 +163,42 @@ def build(self, active_operation_idx=None, use_live_connection=None):
 
     @frappe.whitelist()
     def execute(
-        self, active_operation_idx: int | None = None, adhoc_filters: dict | None = None, force: bool = False
+        self,
+        active_operation_idx: int | None = None,
+        adhoc_filters: dict | None = None,
+        force: bool = False,
+        dashboard: str | None = None,
     ):
         with set_adhoc_filters(adhoc_filters):
             ibis_query = self.build(active_operation_idx)
 
-        limit = 100
-        for op in frappe.parse_json(self.operations):
-            if op.get("limit"):
-                limit = op.get("limit")
-                break
-
-        results, time_taken = execute_ibis_query(
+        response = execute_ibis_query(
             ibis_query,
-            limit,
             force=force,
-            cache_expiry=60 * 10,
             reference_doctype=self.doctype,
             reference_name=self.name,
         )
-        results = results.to_dict(orient="records")
 
-        columns = get_columns_from_schema(ibis_query.schema())
-        return {
-            "sql": ibis.to_sql(ibis_query),
-            "columns": columns,
-            "rows": results,
-            "time_taken": time_taken,
-        }
+        if response.status == "completed":
+            self._publish_query_complete(response.key, dashboard)
+            response.data = response.data.to_dict(orient="records")
+        elif response.status == "pending":
+            pass
+        elif response.status == "queue_full":
+            pass
+
+        return response
+
+    def _publish_query_complete(self, key, dashboard=None):
+        if not dashboard:
+            return
+        # TODO: this doesn't just emits into the dashboard room
+        frappe.publish_realtime(
+            "insights_query_complete",
+            {"key": key},
+            doctype="Insights Dashboard v3",
+            docname=dashboard,
+        )
 
     @insights_whitelist()
     def format(self, raw_sql: str):
@@ -208,6 +218,7 @@ def get_count(self, active_operation_idx: int | None = None, adhoc_filters: dict
             cache_expiry=60 * 5,
             reference_doctype=self.doctype,
             reference_name=self.name,
+            use_lock=False,
         )
         total_count = count_results.values[0][0]
         return int(total_count)
@@ -225,6 +236,7 @@ def download_results(
             limit=10_00_000,
             reference_doctype=self.doctype,
             reference_name=self.name,
+            use_lock=False,
         )
         if format == "excel":
             output = BytesIO()
@@ -261,6 +273,7 @@ def get_distinct_column_values(
             cache_expiry=24 * 60 * 60,
             reference_doctype=self.doctype,
             reference_name=self.name,
+            use_lock=False,
         )
         return result[column_name].tolist()
 
@@ -281,6 +294,7 @@ def evaluate_alert_expression(self, expression):
             cache=False,
             reference_doctype=self.doctype,
             reference_name=self.name,
+            use_lock=False,
         )
         return bool(len(results))
 
diff --git a/insights/insights/doctype/insights_settings/insights_settings.json b/insights/insights/doctype/insights_settings/insights_settings.json
index b02225b6d..156fe984f 100644
--- a/insights/insights/doctype/insights_settings/insights_settings.json
+++ b/insights/insights/doctype/insights_settings/insights_settings.json
@@ -25,7 +25,8 @@
   "allow_subquery",
   "auto_execute_query",
   "query_result_expiry",
-  "query_result_limit"
+  "query_result_limit",
+  "max_concurrent_queries"
  ],
  "fields": [
   {
@@ -140,12 +141,19 @@
    "fieldname": "max_execution_time",
    "fieldtype": "Int",
    "label": "Max Execution Time (Seconds)"
+  },
+  {
+   "default": "6",
+   "fieldname": "max_concurrent_queries",
+   "fieldtype": "Int",
+   "label": "Max Concurrent Queries",
+   "non_negative": 1
   }
  ],
  "index_web_pages_for_search": 1,
  "issingle": 1,
  "links": [],
- "modified": "2025-01-14 13:17:25.798141",
+ "modified": "2026-02-14 10:11:05.771022",
  "modified_by": "Administrator",
  "module": "Insights",
  "name": "Insights Settings",
@@ -183,4 +191,4 @@
  "sort_order": "DESC",
  "states": [],
  "track_changes": 1
-}
\ No newline at end of file
+}
diff --git a/insights/insights/doctype/insights_settings/insights_settings.py b/insights/insights/doctype/insights_settings/insights_settings.py
index 02d2bd1be..fb0f74171 100644
--- a/insights/insights/doctype/insights_settings/insights_settings.py
+++ b/insights/insights/doctype/insights_settings/insights_settings.py
@@ -24,6 +24,7 @@ class InsightsSettings(Document):
         enable_data_store: DF.Check
         enable_permissions: DF.Check
         fiscal_year_start: DF.Date | None
+        max_concurrent_queries: DF.Int
         max_execution_time: DF.Int
         max_memory_usage: DF.Int
         max_records_to_sync: DF.Int
@@ -52,6 +53,8 @@ def update_settings(self, settings: dict | str):
             self.query_result_limit = settings.query_result_limit
         if hasattr(settings, "allow_subquery"):
             self.allow_subquery = settings.allow_subquery
+        if hasattr(settings, "max_concurrent_queries"):
+            self.max_concurrent_queries = settings.max_concurrent_queries
         if hasattr(settings, "telegram_api_token"):
             self.telegram_api_token = settings.telegram_api_token
         self.save()
diff --git a/insights/tests/test_query_locking.py b/insights/tests/test_query_locking.py
new file mode 100644
index 000000000..00b29f21a
--- /dev/null
+++ b/insights/tests/test_query_locking.py
@@ -0,0 +1,274 @@
+# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+import frappe
+import pandas as pd
+
+from insights.insights.doctype.insights_data_source_v3.ibis_utils import (
+    DEFAULT_MAX_CONCURRENT_QUERIES,
+    LOCK_TIMEOUT,
+    QUERY_LOCK_PREFIX,
+    cache_results,
+    execute_with_lock,
+    get_cached_results,
+    get_max_concurrent_queries,
+    get_pending_query_result,
+    has_cached_results,
+    is_query_executing,
+    release_lock,
+    release_semaphore,
+    try_acquire_lock,
+    try_acquire_semaphore,
+)
+
+
+class TestQueryLocking(unittest.TestCase):
+    def setUp(self):
+        self.test_cache_key = f"test_lock_{frappe.generate_hash(length=8)}"
+        self.lock_key = self.test_cache_key
+        self.cleanup_test_keys()
+
+    def tearDown(self):
+        self.cleanup_test_keys()
+
+    def cleanup_test_keys(self):
+        try:
+            cache = frappe.cache()
+            cache.delete(cache.make_key(self.lock_key))
+            results_key = "insights:query_results:" + self.test_cache_key
+            cache.delete(cache.make_key(results_key))
+        except Exception:
+            pass
+
+    def test_acquire_lock_success(self):
+        acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(acquired, "Should acquire lock when not held")
+        release_lock(self.lock_key)
+
+    def test_acquire_lock_fails_when_held(self):
+        first_acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(first_acquired, "First acquisition should succeed")
+
+        second_acquired = try_acquire_lock(self.lock_key)
+        self.assertFalse(second_acquired, "Second acquisition should fail")
+
+        release_lock(self.lock_key)
+
+    def test_release_lock(self):
+        try_acquire_lock(self.lock_key)
+        release_lock(self.lock_key)
+
+        acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(acquired, "Should acquire lock after release")
+        release_lock(self.lock_key)
+
+    def test_is_query_executing(self):
+        self.assertFalse(is_query_executing(self.test_cache_key), "Should not be executing initially")
+
+        acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(acquired, "Lock should be acquired")
+
+        # is_query_executing checks the raw key, but lock uses prefixed key
+        # so this checks the cache_key without prefix — matches how it's used in get_pending_query_result
+        executing = is_query_executing(self.lock_key)
+        self.assertTrue(executing, "Should be executing after lock acquired")
+
+        release_lock(self.lock_key)
+
+        self.assertFalse(is_query_executing(self.lock_key), "Should not be executing after lock released")
+
+    def test_semaphore_acquire_success(self):
+        slot = try_acquire_semaphore()
+        self.assertIsNotNone(slot, "Should acquire semaphore slot")
+        release_semaphore()
+
+    def test_semaphore_limit(self):
+        max_queries = get_max_concurrent_queries()
+        acquired_slots = []
+
+        for _i in range(max_queries):
+            slot = try_acquire_semaphore()
+            if slot is not None:
+                acquired_slots.append(slot)
+        extra_slot = try_acquire_semaphore()
+        self.assertIsNone(extra_slot, f"Should not acquire more than {max_queries} slots")
+
+        for _ in acquired_slots:
+            release_semaphore()
+
+    def test_semaphore_release(self):
+        max_queries = get_max_concurrent_queries()
+
+        for _ in range(max_queries):
+            try_acquire_semaphore()
+
+        self.assertIsNone(try_acquire_semaphore())
+
+        release_semaphore()
+        slot = try_acquire_semaphore()
+        self.assertIsNotNone(slot, "Should acquire after release")
+
+        for _ in range(max_queries):
+            release_semaphore()
+
+    def test_cache_results(self):
+        test_data = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
+
+        cache_results(self.test_cache_key, test_data, cache_expiry=60)
+
+        self.assertTrue(has_cached_results(self.test_cache_key), "Should have cached results")
+
+        cached = get_cached_results(self.test_cache_key)
+        self.assertIsNotNone(cached, "Should retrieve cached results")
+        self.assertEqual(len(cached), 3, "Should have 3 rows")
+
+    def test_cache_miss(self):
+        non_existent_key = f"non_existent_{frappe.generate_hash(length=8)}"
+
+        self.assertFalse(
+            has_cached_results(non_existent_key), "Should not have cached results for non-existent key"
+        )
+
+        self.assertIsNone(get_cached_results(non_existent_key), "Should return None for cache miss")
+
+    def test_pending_query_result_executing(self):
+        acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(acquired, "Should acquire lock for test")
+
+        # get_pending_query_result checks is_query_executing with the raw cache_key
+        # but the lock is on lock_key (with prefix), so this tests the actual flow
+        result = get_pending_query_result(self.lock_key)
+        self.assertEqual(result["status"], "pending", "Should return pending status")
+
+        release_lock(self.lock_key)
+
+    def test_pending_query_result_completed(self):
+        test_data = pd.DataFrame({"value": [717]})
+        cache_results(self.test_cache_key, test_data, cache_expiry=60)
+
+        result = get_pending_query_result(self.test_cache_key)
+        self.assertEqual(result["status"], "completed", "Should return completed status")
+        self.assertIsNotNone(result["result"], "Should have result data")
+
+    def test_pending_query_result_not_found(self):
+        non_existent_key = f"non_existent_{frappe.generate_hash(length=8)}"
+
+        result = get_pending_query_result(non_existent_key)
+        self.assertEqual(result["status"], "not_found", "Should return not_found status")
+
+    def test_execute_with_lock_returns_pending_when_locked(self):
+        try_acquire_lock(self.lock_key)
+
+        mock_query = MagicMock()
+
+        result, time_taken = execute_with_lock(
+            mock_query,
+            "SELECT 1",
+            self.test_cache_key,
+            cache=True,
+            cache_expiry=60,
+            force=False,
+            reference_name="test",
+        )
+
+        self.assertIsInstance(result, dict, "Should return dict")
+        self.assertEqual(result["status"], "pending", "Should return pending status")
+        self.assertEqual(result["cache_key"], self.test_cache_key)
+
+        release_lock(self.lock_key)
+
+    def test_execute_with_lock_releases_lock_on_completion(self):
+        mock_query = MagicMock()
+        mock_query.execute.return_value = pd.DataFrame({"result": [1]})
+
+        with patch("insights.insights.doctype.insights_data_source_v3.ibis_utils.create_execution_log"):
+            execute_with_lock(
+                mock_query,
+                "SELECT 1",
+                self.test_cache_key,
+                cache=True,
+                cache_expiry=60,
+                force=False,
+                reference_name="test",
+            )
+
+        self.assertFalse(
+            is_query_executing(self.test_cache_key),
+            "Lock should be released after execution",
+        )
+
+    def test_execute_with_lock_releases_lock_on_error(self):
+        mock_query = MagicMock()
+        mock_query.execute.side_effect = Exception("Test error")
+
+        with self.assertRaises(Exception):
+            with patch("insights.insights.doctype.insights_data_source_v3.ibis_utils.create_execution_log"):
+                execute_with_lock(
+                    mock_query,
+                    "SELECT 1",
+                    self.test_cache_key,
+                    cache=True,
+                    cache_expiry=60,
+                    force=False,
+                    reference_name="test",
+                )
+
+        self.assertFalse(
+            is_query_executing(self.test_cache_key),
+            "Lock should be released even after error",
+        )
+
+    def test_coalescing_flow(self):
+        first_acquired = try_acquire_lock(self.lock_key)
+        self.assertTrue(first_acquired, "First request should acquire lock")
+
+        self.assertTrue(is_query_executing(self.lock_key), "Query should be marked as executing")
+
+        result = get_pending_query_result(self.lock_key)
+        self.assertEqual(result["status"], "pending", "Second request should get pending")
+
+        test_data = pd.DataFrame({"value": [717]})
+        cache_results(self.test_cache_key, test_data, cache_expiry=60)
+        release_lock(self.lock_key)
+
+        result = get_pending_query_result(self.test_cache_key)
+        self.assertEqual(result["status"], "completed", "Should get completed after cache")
+        self.assertEqual(len(result["result"]), 1)
+        self.assertEqual(result["result"]["value"].iloc[0], 717)
+
+    def test_redis_failure_fails_open(self):
+        original_cache = frappe.cache
+
+        def mock_cache():
+            mock = MagicMock()
+            mock.set.side_effect = Exception("Redis connection failed")
+            return mock
+
+        try:
+            frappe.cache = mock_cache
+            acquired = try_acquire_lock("test_lock_failure")
+            self.assertTrue(acquired, "Should fail open and allow execution")
+        finally:
+            frappe.cache = original_cache
+
+
+class TestQueryLockingConstants(unittest.TestCase):
+    def test_lock_timeout_reasonable(self):
+        self.assertGreaterEqual(LOCK_TIMEOUT, 60, "Lock timeout should be at least 60 seconds")
+        self.assertLessEqual(LOCK_TIMEOUT, 600, "Lock timeout should be at most 10 minutes")
+
+    def test_default_max_concurrent_queries_reasonable(self):
+        self.assertGreaterEqual(DEFAULT_MAX_CONCURRENT_QUERIES, 1)
+        self.assertLessEqual(DEFAULT_MAX_CONCURRENT_QUERIES, 18)
+
+    def test_lock_prefix_namespaced(self):
+        self.assertTrue(
+            QUERY_LOCK_PREFIX.startswith("insights"), "Lock prefix should be namespaced to insights"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()