diff --git a/sqlalchemy_kusto/dialect_kql.py b/sqlalchemy_kusto/dialect_kql.py
index 5be6be3..cbc82d4 100644
--- a/sqlalchemy_kusto/dialect_kql.py
+++ b/sqlalchemy_kusto/dialect_kql.py
@@ -66,6 +66,58 @@
     "variancep",
 }
 AGGREGATE_PATTERN = r"(\w+)\s*\(\s*(DISTINCT|distinct\s*)?\(?\s*(\*|\[?\"?\'?\w+\"?\]?)\s*(,.+)*\)?\s*\)"
+# Pre-compiled regex for aggregate function matching (performance optimization).
+# Compiled once at module load to avoid recompiling on every call, which significantly
+# improves performance for query-heavy workloads.
+KQL_AGG_PATTERN = re.compile(
+    r"\b(" + "|".join(kql_aggregates) + r")\s*\(", re.IGNORECASE
+)
+
+
+class _ParseState:
+    """Tracks parsing state while scanning through text."""
+
+    __slots__ = ("in_double_quote", "in_single_quote", "in_bracket", "paren_depth")
+
+    def __init__(self):
+        self.in_double_quote = False
+        self.in_single_quote = False
+        self.in_bracket = False
+        self.paren_depth = 0
+
+    def update(self, ch: str, prev_ch: str | None) -> None:
+        """Update state based on current and previous character."""
+        # Handle quotes (only if not escaped and not in conflicting context)
+        if (
+            ch == '"'
+            and prev_ch != "\\"
+            and not self.in_single_quote
+            and not self.in_bracket
+        ):
+            self.in_double_quote = not self.in_double_quote
+        elif (
+            ch == "'"
+            and prev_ch != "\\"
+            and not self.in_double_quote
+            and not self.in_bracket
+        ):
+            self.in_single_quote = not self.in_single_quote
+        # Handle brackets and parens (only if not in quotes)
+        elif not self.in_double_quote and not self.in_single_quote:
+            if ch == "[":
+                self.in_bracket = True
+            elif ch == "]":
+                self.in_bracket = False
+            elif not self.in_bracket:
+                if ch == "(":
+                    self.paren_depth += 1
+                elif ch == ")":
+                    self.paren_depth -= 1
+
+    @property
+    def in_quotes_or_brackets(self) -> bool:
+        """Check if currently inside quotes or brackets."""
+        return self.in_double_quote or self.in_single_quote or self.in_bracket
 
 
 class UniversalSet:
@@ -89,6 +141,149 @@ class KustoKqlCompiler(compiler.SQLCompiler):
     visit_sequence = None
     sort_with_clause_parts = 2
 
+    @staticmethod
+    def _find_top_level_operator(text: str, operator: str) -> int:
+        """Find position of operator at depth 0 (not inside quotes, brackets, or parens).
+
+        Args:
+            text: The string to search in
+            operator: The single-character operator to find (e.g., '+', '-', '*', '/')
+
+        Returns:
+            The position of the operator at depth 0, or -1 if not found.
+            Returns -1 when the operator only appears inside quotes, brackets, or nested parens.
+        """
+        state = _ParseState()
+        for i, ch in enumerate(text):
+            if (
+                ch == operator
+                and state.paren_depth == 0
+                and not state.in_quotes_or_brackets
+            ):
+                return i
+            state.update(ch, text[i - 1] if i > 0 else None)
+        return -1
+
+    @staticmethod
+    def _is_inside_quotes_or_brackets(text: str, pos: int) -> bool:
+        """Check if a position in text is inside quotes or brackets."""
+        if pos >= len(text):
+            return False
+
+        state = _ParseState()
+        for i in range(pos):
+            state.update(text[i], text[i - 1] if i > 0 else None)
+        return state.in_quotes_or_brackets
+
+    @staticmethod
+    def _find_matching_paren(text: str, start_pos: int) -> int:
+        """Find the matching closing parenthesis for an opening paren at start_pos."""
+        if start_pos >= len(text) or text[start_pos] != "(":
+            return -1
+
+        state = _ParseState()
+        state.paren_depth = 1  # Start with depth 1 since we're at opening paren
+
+        for i in range(start_pos + 1, len(text)):
+            ch = text[i]
+            state.update(ch, text[i - 1] if i > 0 else None)
+            if state.paren_depth == 0:
+                return i
+        return -1
+
+    @staticmethod
+    def _has_operators_outside_quotes(expr: str) -> bool:
+        """Check if expression has arithmetic operators outside of quoted strings and brackets."""
+        return any(
+            KustoKqlCompiler._find_top_level_operator(expr, op) != -1 for op in "+-*/"
+        )
+
+    @staticmethod
+    def _count_outer_parens(text: str) -> tuple[int, str]:
+        """Count and strip outer parentheses from text. Returns (count, stripped_text)."""
+        text = text.strip()
+        count = 0
+        while len(text) > 1 and text[0] == "(" and text[-1] == ")":
+            depth = 0
+            for ch in text[:-1]:  # Scan all but last char
+                depth += (ch == "(") - (ch == ")")
+                if depth == 0:
+                    return count, text  # First '(' closed before end
+            count += 1
+            text = text[1:-1].strip()
+        return count, text
+
+    @staticmethod
+    def _extract_and_replace_aggregates(
+        expr: str, measure_name: str, existing_aggs: dict[str, str] | None = None
+    ) -> tuple[str, list[tuple[str, str]]]:
+        """Extract aggregate functions from an expression and replace with references.
+
+        Args:
+            expr: The expression to process (may contain aggregates, operators)
+            measure_name: Name of the parent measure (used for generating ref names)
+            existing_aggs: Dict mapping kql_agg (lowercase) -> ref_name for reuse.
+
+        Returns:
+            A tuple of:
+            - modified expression with aggregates replaced by references like ["__measure_1"]
+            - list of (ref_name, kql_aggregate) tuples to add to summarize (only NEW ones)
+        """
+        if existing_aggs is None:
+            existing_aggs = {}
+
+        new_aggregates: list[tuple[str, str]] = []
+        agg_counter = 0
+
+        # Collect replacements: (start, end, ref_name)
+        replacements: list[tuple[int, int, str]] = []
+
+        for match in KQL_AGG_PATTERN.finditer(expr):
+            start = match.start()
+            paren_end = KustoKqlCompiler._find_matching_paren(expr, match.end() - 1)
+            kql_agg = (
+                KustoKqlCompiler._extract_maybe_agg_column_parts(
+                    expr[start : paren_end + 1]
+                )
+                if paren_end != -1
+                else None
+            )
+
+            # Skip invalid matches
+            if (
+                KustoKqlCompiler._is_inside_quotes_or_brackets(expr, start)
+                or paren_end == -1
+                or not kql_agg
+            ):
+                continue
+
+            # Reuse existing aggregate or create new one
+            if kql_agg.lower() in existing_aggs:
+                ref_name = existing_aggs[kql_agg.lower()]
+            else:
+                agg_counter += 1
+                clean_name = measure_name.strip('[]"')
+                ref_name = f'["__{clean_name}_{agg_counter}"]'
+                existing_aggs[kql_agg.lower()] = ref_name
+                new_aggregates.append((ref_name, kql_agg))
+
+            replacements.append((start, paren_end + 1, ref_name))
+
+        # Apply replacements from right to left so positions stay valid
+        result = expr
+        for start, end, ref_name in reversed(replacements):
+            result = result[:start] + ref_name + result[end:]
+
+        return result, new_aggregates
+
+    @staticmethod
+    def _contains_aggregate_function(expr: str) -> bool:
+        """Check if expression contains an aggregate function call."""
+        for match in KQL_AGG_PATTERN.finditer(expr):
+            if not KustoKqlCompiler._is_inside_quotes_or_brackets(expr, match.start()):
+                return True
+        return False
+
     def visit_select(
         self,
         select_stmt: selectable.Select,
@@ -107,7 +302,7 @@ def visit_select(
         from_object = select_stmt.get_final_froms()[0]
         if hasattr(from_object, "element"):
             query = self._get_most_inner_element(from_object.element)
-            (main, lets) = self._extract_let_statements(query.text)
+            main, lets = self._extract_let_statements(query.text)
             compiled_query_lines.extend(lets)
             compiled_query_lines.append(
                 f"let {from_object.name} = ({self._convert_schema_in_statement(main)});"
@@ -142,9 +337,15 @@ def visit_select(
                 )
                 compiled_query_lines.append(f"| where {converted_where_clause}")
 
+        # Add summarize first if it exists
+        if "summarize" in projections_parts_dict:
+            compiled_query_lines.append(projections_parts_dict.pop("summarize"))
+
+        # Then add extend after summarize
         if "extend" in projections_parts_dict:
             compiled_query_lines.append(projections_parts_dict.pop("extend"))
 
+        # Add remaining parts (project, sort)
         for statement_part in projections_parts_dict.values():
             if statement_part:
                 compiled_query_lines.append(statement_part)
@@ -212,31 +413,63 @@ def _get_projection_or_summarize(self, select: selectable.Select) -> dict[str, s
         #                |
         #                N---> Add to projection
         if columns is not None:
-            summarize_columns = set()
-            extend_columns = set()
+            summarize_columns = []
+            extend_columns = []
             projection_columns = []
+            existing_aggs: dict[str, str] = {}
             for column in [c for c in columns if c.name != "*"]:
                 column_name, column_alias = self._extract_column_name_and_alias(column)
                 column_alias = self._escape_and_quote_columns(column_alias, True)
+                column_name = re.sub(
+                    r'(?:[a-zA-Z_][a-zA-Z0-9_]*|\["[^"]+"\])\.', "", column_name
+                )  # Remove table qualifiers from column name for processing
                 # Do we have a group by clause ?
                 # Do we have aggregate columns ?
                 kql_agg = self._extract_maybe_agg_column_parts(column_name)
-                if kql_agg:
+                has_inline_aggregates = self._contains_aggregate_function(column_name)
+
+                # Case 1: Simple aggregate (e.g., count(), sum(col))
+                if kql_agg and not self._has_operators_outside_quotes(column_name):
                     has_aggregates = True
-                    summarize_columns.add(
-                        self._build_column_projection(kql_agg, column_alias)
+                    summarize_entry = self._build_column_projection(
+                        kql_agg, column_alias
                     )
+                    summarize_columns.append(summarize_entry)
+                    projection_columns.append(column_alias)
+                    if column_alias:
+                        existing_aggs[kql_agg.lower()] = column_alias
+
+                # Case 2 & 3: Expressions with aggregates or aliased columns (both go to extend)
                 # No group by clause
                 # Do the columns have aliases ?
                 # Add additional and to handle case where : SELECT column_name as column_name
-                elif column_alias and column_alias != column_name:
-                    extend_columns.add(
-                        self._build_column_projection(column_name, column_alias, True)
-                    )
-                if column_alias:
-                    projection_columns.append(
-                        self._escape_and_quote_columns(column_alias, True)
-                    )
+                elif has_inline_aggregates or (
+                    column_alias
+                    and column_alias != self._escape_and_quote_columns(column_name)
+                ):
+                    # If contains aggregates, extract them first
+                    if has_inline_aggregates:
+                        has_aggregates = True
+                        column_name, extracted_aggs = (
+                            self._extract_and_replace_aggregates(
+                                column_name, column_alias or "expr", existing_aggs
+                            )
+                        )
+
+                        # Add extracted aggregates to summarize
+                        for ref_name, kql_agg_extracted in extracted_aggs:
+                            summarize_entry = self._build_column_projection(
+                                kql_agg_extracted, ref_name
+                            )
+                            summarize_columns.append(summarize_entry)
+
+                    # Build extend entry (common for both cases)
+                    escaped_expr = self._escape_and_quote_columns(column_name)
+                    extend_entry = f"{column_alias} = {escaped_expr}"
+                    extend_columns.append(extend_entry)
+                    projection_columns.append(column_alias)
+
+                # Case 4: Simple column reference
                 else:
                     projection_columns.append(
                         self._escape_and_quote_columns(column_name)
@@ -252,7 +485,8 @@ def _get_projection_or_summarize(self, select: selectable.Select) -> dict[str, s
                         f"{summarize_statement} by {', '.join(by_columns)}"
                     )
             if extend_columns:
-                extend_statement = f"| extend {', '.join(sorted(extend_columns))}"
+                extend_statement = f"| extend {', '.join(extend_columns)}"
+
             project_statement = (
                 f"| project {', '.join(projection_columns)}"
                 if projection_columns
@@ -356,26 +590,36 @@ def _escape_and_quote_columns(name: str | None, is_alias=False) -> str:
             or KustoKqlCompiler._is_number_literal(name)
         ) and not is_alias:
             return name
-        if name.startswith('"') and name.endswith('"'):
-            name = name[1:-1]
         # First, check if the name is already wrapped in ["ColumnName"] (escaped format)
         if name.startswith('["') and name.endswith('"]'):
             return name  # Return as is if already properly escaped
-        # Remove surrounding spaces
-        # Handle mathematical operations (wrap only the column part before operators)
-        # Find the position of the first operator or space that separates the column name
+        # Handle arithmetic expressions by recursively processing operands
         if not is_alias:
+            outer_paren_count, inner = KustoKqlCompiler._count_outer_parens(name)
             for operator in ["/", "+", "-", "*"]:
-                if operator in name:
-                    # Split the name at the first operator and wrap the left part
-                    parts = name.split(operator, 1)
-                    # Remove quotes if they exist at the edges
-                    col_part = parts[0].strip()
-                    if col_part.startswith('"') and col_part.endswith('"'):
-                        col_part = col_part[1:-1].strip()
-                    col_part = col_part.replace('"', '\\"')
-                    return f'["{col_part}"] {operator} {parts[1].strip()}'  # Wrap the column part
-        # No operators found, just wrap the entire name
+                pos = KustoKqlCompiler._find_top_level_operator(inner, operator)
+                if pos != -1:
+                    left = KustoKqlCompiler._escape_and_quote_columns(
+                        inner[:pos].strip()
+                    )
+                    right = KustoKqlCompiler._escape_and_quote_columns(
+                        inner[pos + 1 :].strip()
+                    )
+                    return (
+                        "(" * outer_paren_count
+                        + left
+                        + " "
+                        + operator
+                        + " "
+                        + right
+                        + ")" * outer_paren_count
+                    )
+            # No operators - just process inner content (don't re-add unnecessary parens)
+            if outer_paren_count > 0:
+                return KustoKqlCompiler._escape_and_quote_columns(inner)
+        # No operators found - strip surrounding quotes if present, then wrap
+        if name.startswith('"') and name.endswith('"'):
+            name = name[1:-1]
         name = name.replace('"', '\\"')
         return f'["{name}"]'
 
@@ -547,7 +791,7 @@ def _is_kql_function(name: str) -> bool:
 
     @staticmethod
     def _is_number_literal(s: str) -> bool:
-        pattern = r"^[0-9]+$"
+        pattern = r"^\d+(\.\d+)?$"
         return bool(re.match(pattern, s))
 
     def _get_most_inner_element(self, clause):
diff --git a/tests/integration/test_dialect_kql.py b/tests/integration/test_dialect_kql.py
index 5410337..028d0da 100644
--- a/tests/integration/test_dialect_kql.py
+++ b/tests/integration/test_dialect_kql.py
@@ -202,6 +202,94 @@ def test_date_bin_ops(test_label, group_fn, temp_table_name, expected, compare_d
         assert actual_result == expected_records
 
 
+def test_parentheses_preserved_in_expression(temp_table_name):
+    table = Table(
+        temp_table_name,
+        metadata,
+        Column("Text", String),
+    )
+    measure = literal_column('(("Text"))').label("Measure 1")
+    query = (
+        session.query(measure)
+        .select_from(table)
+        .order_by(literal_column('"Measure 1"'))
+    )
+    query_compiled = str(query.statement.compile(kql_engine)).replace("\n", "")
+
+    assert "((" in query_compiled
+    assert "))" in query_compiled
+    assert '["Text"]' in query_compiled
+
+    with kql_engine.connect() as connection:
+        result = connection.execute(text(query_compiled))
+        values = [row[0] for row in result.fetchall()]
+        expected_count = 9
+        assert len(values) == expected_count
+        assert set(values) == {"value_0", "value_1"}
+
+
+def test_quoted_identifier_converted_to_kql(temp_table_name):
+    table = Table(
+        temp_table_name,
+        metadata,
+        Column("Text", String),
+    )
+    quoted_text = literal_column("Text").label("QuotedText")
+    query = session.query(quoted_text).select_from(table).order_by(text("QuotedText"))
+    query_compiled = str(query.statement.compile(kql_engine)).replace("\n", "")
+
+    assert '["Text"]' in query_compiled
+
+    with kql_engine.connect() as connection:
+        result = connection.execute(text(query_compiled))
+        values = [row[0] for row in result.fetchall()]
+        expected_count = 9
+        assert len(values) == expected_count
+        assert set(values) == {"value_0", "value_1"}
+
+
+def test_uppercase_functions_lowercased(temp_table_name):
+    table = Table(
+        temp_table_name,
+        metadata,
+    )
+    query = session.query(func.COUNT(text("Id")).label("tag_count")).select_from(table)
+    query_compiled = str(query.statement.compile(kql_engine)).replace("\n", "")
+
+    assert "count(" in query_compiled
+    assert "COUNT(" not in query_compiled
+
+    with kql_engine.connect() as connection:
+        result = connection.execute(text(query_compiled))
+        assert {row[0] for row in result.fetchall()} == {9}
+
+
+def test_extend_after_summarize_for_calculated_measures(temp_table_name):
+    table = Table(
+        temp_table_name,
+        metadata,
+    )
+    measure_1 = func.COUNT(text("Id")).label("Measure 1")
+    measure_2 = literal_column('"Measure 1" * 2').label("Measure 2")
+    query = session.query(measure_1, measure_2).select_from(table)
+    query_compiled = str(query.statement.compile(kql_engine)).replace("\n", "")
+
+    summarize_index = query_compiled.find("| summarize")
+    extend_index = query_compiled.find("| extend")
+    assert summarize_index != -1
+    assert extend_index != -1
+    assert summarize_index < extend_index
+
+    with kql_engine.connect() as connection:
+        result = connection.execute(text(query_compiled))
+        row = result.fetchone()
+        assert row is not None
+        expected_count = 9
+        expected_double_count = 18
+        assert int(row[0]) == expected_count
+        assert int(row[1]) == expected_double_count
+
+
 def get_kcsb():
     return (
         KustoConnectionStringBuilder.with_az_cli_authentication(KUSTO_URL)
diff --git a/tests/unit/test_dialect_kql.py b/tests/unit/test_dialect_kql.py
index 04b8fc0..ff92bf6 100644
--- a/tests/unit/test_dialect_kql.py
+++ b/tests/unit/test_dialect_kql.py
@@ -173,11 +173,12 @@ def test_group_by_text():
     query_compiled = str(
         query.compile(engine, compile_kwargs={"literal_binds": True})
     ).replace("\n", "")
-    # raw query text from query
+    # raw query text from query - order matches column appearance
     query_expected = (
-        '["ActiveUsersLastMonth"]| extend ["ActiveUserMetric"] = ["ActiveUsers"], '
-        '["EventInfo_Time"] = ["EventInfo_Time"] / time(1d)'
+        '["ActiveUsersLastMonth"]'
         '| summarize   by ["EventInfo_Time"] / time(1d)'
+        '| extend ["EventInfo_Time"] = ["EventInfo_Time"] / time(1d), '
+        '["ActiveUserMetric"] = ["ActiveUsers"]'
         '| project ["EventInfo_Time"], ["ActiveUserMetric"]'
         '| order by ["ActiveUserMetric"] desc'
     )
@@ -201,12 +202,13 @@ def test_function_text(f: str, expected: str):
     query_compiled = str(
         query.compile(engine, compile_kwargs={"literal_binds": True})
     ).replace("\n", "")
+    # Order matches column appearance in select
     query_expected = (
         '["ActiveUsersLastMonth"]'
-        '| extend ["ActiveUserMetric"] = ["ActiveUsers"], '
-        '["EventInfo_Time"] = '
+        '| extend ["EventInfo_Time"] = '
         + expected
-        + '| project ["EventInfo_Time"], ["ActiveUserMetric"]'
+        + ', ["ActiveUserMetric"] = ["ActiveUsers"]'
+        '| project ["EventInfo_Time"], ["ActiveUserMetric"]'
     )
     assert query_compiled == query_expected
 
@@ -214,6 +216,7 @@ def test_function_text(f: str, expected: str):
 def test_group_by_text_vaccine_dataset():
     # SQL: SELECT country_name AS country_name FROM superset."CovidVaccineData" GROUP BY country_name
     # ORDER BY country_name ASC - this is a simple query to get distinct country names
+    # Note: When alias = column name, no extend is needed
     query = (
         select([literal_column("country_name").label("country_name")])
         .select_from(text('superset."CovidVaccineData"'))
@@ -224,20 +227,18 @@ def test_group_by_text_vaccine_dataset():
         query.compile(engine, compile_kwargs={"literal_binds": True})
     ).replace("\n", "")
     query_expected = (
-        'database("superset").["CovidVaccineData"]| '
-        'extend ["country_name"] = ["country_name"]| '
-        'summarize   by ["country_name"]| '
-        'project ["country_name"]| order by ["country_name"] asc'
+        'database("superset").["CovidVaccineData"]'
+        '| summarize   by ["country_name"]'
+        '| project ["country_name"]'
+        '| order by ["country_name"] asc'
     )
     assert query_compiled == query_expected
 
 
 def test_is_kql_function():
-    assert KustoKqlCompiler._is_kql_function(
-        """case(Size <= 3, "Small",
+    assert KustoKqlCompiler._is_kql_function("""case(Size <= 3, "Small",
                        Size <= 10, "Medium",
-                       "Large")"""
-    )
+                       "Large")""")
     assert KustoKqlCompiler._is_kql_function("""bin(time(16d), 7d)""")
     assert KustoKqlCompiler._is_kql_function(
         """iff((EventType in ("Heavy Rain", "Flash Flood", "Flood")), "Rain event", "Not rain event")"""
@@ -328,8 +329,8 @@ def test_distinct_count_by_text():
     # raw query text from query
     query_expected = (
         '["ActiveUsersLastMonth"]'
-        '| extend ["EventInfo_Time"] = ["EventInfo_Time"] / time(1d)'
         '| summarize ["DistinctUsers"] = dcount(["ActiveUsers"])  by ["EventInfo_Time"] / time(1d)'
+        '| extend ["EventInfo_Time"] = ["EventInfo_Time"] / time(1d)'
         '| project ["EventInfo_Time"], ["DistinctUsers"]'
         '| order by ["ActiveUserMetric"] desc'
     )
@@ -354,8 +355,8 @@ def test_distinct_count_alt_by_text():
     # raw query text from query
     query_expected = (
         '["ActiveUsersLastMonth"]'
-        '| extend ["EventInfo_Time"] = ["EventInfo_Time"] / time(1d)'
         '| summarize ["DistinctUsers"] = dcount(["ActiveUsers"])  by ["EventInfo_Time"] / time(1d)'
+        '| extend ["EventInfo_Time"] = ["EventInfo_Time"] / time(1d)'
         '| project ["EventInfo_Time"], ["DistinctUsers"]'
         '| order by ["ActiveUserMetric"] desc'
     )
@@ -549,6 +550,718 @@ def test_match_aggregates(column_name: str, expected_aggregate: str):
         assert kql_agg is None
 
 
+def test_escape_and_quote_columns_with_two_quoted_measures():
+    """Test that two quoted measure names with operator are properly escaped.
+
+    e.g. "Measure 1" + "Measure 2" --> ["Measure 1"] + ["Measure 2"]
+    """
+    result = KustoKqlCompiler._escape_and_quote_columns('"Measure 1" + "Measure 2"')
+    assert result == '["Measure 1"] + ["Measure 2"]'
+
+
+def test_escape_and_quote_columns_preserves_already_bracketed():
+    """Test that already-bracketed columns are not double-converted."""
+    result = KustoKqlCompiler._escape_and_quote_columns('["Measure 1"]')
+    assert result == '["Measure 1"]'
+
+
+class TestCalculatedMeasuresWithParentheses:
+    """Tests for calculated measures with parentheses support."""
+
+    @pytest.fixture
+    def pt_search_table(self):
+        """Table matching the Superset PT_Search_scenario use case."""
+        metadata = MetaData()
+        return Table(
+            "PT_Search_scenario",
+            metadata,
+            Column("UserInfo_Ring", String),
+            Column("UserInfo_Region", String),
+            schema="test_schema",
+        )
+
+    def test_calculated_measure_single_paren(self, pt_search_table):
+        """Test a calculated measure with single parentheses wrapper."""
+        measure_16 = literal_column('("UserInfo_Ring Count")').label("Measure 16")
+
+        query = select(measure_16).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        assert '["Measure 16"]' in compiled
+
+    def test_calculated_measure_double_paren(self, pt_search_table):
+        """Test a calculated measure with double parentheses wrapper.
+
+        Double parens around a single measure reference are stripped since
+        they're unnecessary for precedence.
+        """
+        measure_3 = literal_column('(("Measure 1"))').label("Measure 3")
+
+        query = select(measure_3).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        assert '["Measure 3"]' in compiled
+        # Parens should be stripped for single values (no operators inside)
+        assert '["Measure 1"]' in compiled
+
+    def test_calculated_measure_parens_addition(self, pt_search_table):
+        """Test a calculated measure with parenthesized addition."""
+        measure_11 = literal_column('("Measure 1") + ("Measure 2")').label("Measure 11")
+
+        query = select(measure_11).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        assert '["Measure 11"]' in compiled
+        assert "+" in compiled
+
+    def test_calculated_measure_complex_expression(self, pt_search_table):
+        """Test a complex calculated measure with nested parens and multiplication."""
+        measure_8 = literal_column(
+            '("UserInfo_Ring Count" + "UserInfo_Region Count") * 2'
+        ).label("Measure 8")
+
+        query = select(measure_8).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        assert '["Measure 8"]' in compiled
+        assert "* 2" in compiled
+        assert "+" in compiled
+
+    def test_no_double_bracketing(self, pt_search_table):
+        """Test that there's no double bracketing like [["col"]]."""
+        measure = literal_column('"UserInfo_Ring Count"').label("Test Measure")
+
+        query = select(measure).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        # Should not have double brackets
+        assert '[["' not in compiled
+        assert '"]]' not in compiled
+
+    def test_wrapped_aggregate_extracted_correctly(self, pt_search_table):
+        """Test that aggregates wrapped in parens (like ((COUNT(col)))) are extracted correctly."""
+        measure_4 = literal_column("((COUNT(UserInfo_Ring)))").label("Measure 4")
+
+        query = select(measure_4).select_from(pt_search_table)
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        # Should have summarize with the aggregate
+        assert "summarize" in compiled
+
+        # Find the extend part
+        extend_idx = compiled.find("extend")
+        if extend_idx != -1:
+            project_idx = compiled.find("| project")
+            extend_part = (
+                compiled[extend_idx:project_idx]
+                if project_idx != -1
+                else compiled[extend_idx:]
+            )
+
+            # Should NOT have COUNT() in extend
+            assert "COUNT(" not in extend_part
+            assert "count(" not in extend_part
+            # Should have a reference
+            assert '["Measure 4"]' in extend_part
+
+    def test_floating_point_numbers(self, pt_search_table):
+        """Test that floating point numbers are preserved correctly."""
+        measure_1 = literal_column("count()").label("Measure 1")
+        measure_2 = literal_column('"Measure 1" * 0.5').label("Measure 2")
+        measure_3 = literal_column('"Measure 1" * 1.25').label("Measure 3")
+        measure_4 = literal_column('"Measure 1" / 0.1').label("Measure 4")
+
+        query = select(measure_1, measure_2, measure_3, measure_4).select_from(
+            pt_search_table
+        )
+        compiled = str(query.compile(engine, compile_kwargs={"literal_binds": True}))
+
+        # Floating point numbers should be preserved, not wrapped in brackets
+        assert "* 0.5" in compiled
+        assert "* 1.25" in compiled
+        assert "/ 0.1" in compiled
+        # Should NOT have bracketed numbers
+        assert '["0.5"]' not in compiled
+        assert '["1.25"]' not in compiled
+        assert '["0.1"]' not in compiled
+
+
+def test_calculated_measure_with_two_adhoc_measures():
+    """Test calculated measure referencing two ad hoc measures.
+
+    Measure 3 = "Measure 1" + "Measure 2" should compile to (["Measure 1"]) + (["Measure 2"])
+    Parentheses are added for arithmetic precedence clarity.
+    """
+    measure_3 = literal_column('"Measure 1" + "Measure 2"').label("Measure 3")
+    query = select([measure_3]).select_from(text("SalesData"))
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", "")
+    query_expected = (
+        '["SalesData"]'
+        '| extend ["Measure 3"] = ["Measure 1"] + ["Measure 2"]'
+        '| project ["Measure 3"]'
+    )
+    assert query_compiled == query_expected
+
+
+def test_escape_and_quote_columns_measure_with_constant():
+    """Test that measure with operator and constant is properly escaped.
+
+    e.g. "Measure 1" * 2 --> ["Measure 1"] * 2
+    """
+    result = KustoKqlCompiler._escape_and_quote_columns('"Measure 1" * 2')
+    assert result == '["Measure 1"] * 2'
+
+
+def test_escape_and_quote_columns_measure_with_operator_in_name():
+    """Test that measure names containing operators are properly escaped.
+
+    e.g. "Measure 1-2" --> ["Measure 1-2"] (not split as ["Measure 1"] - ["2"])
+    """
+    result = KustoKqlCompiler._escape_and_quote_columns('"Measure 1-2"')
+    assert result == '["Measure 1-2"]'
+
+
+def test_is_number_literal():
+    """Test _is_number_literal correctly identifies numeric literals."""
+    # Should match: integers and decimals with digits on both sides of decimal
+    assert KustoKqlCompiler._is_number_literal("5") is True
+    assert KustoKqlCompiler._is_number_literal("123") is True
+    assert KustoKqlCompiler._is_number_literal("0") is True
+    assert KustoKqlCompiler._is_number_literal("0.5") is True
+    assert KustoKqlCompiler._is_number_literal("5.0") is True
+    assert KustoKqlCompiler._is_number_literal("123.456") is True
+
+    # Should NOT match: trailing decimal, leading decimal, scientific notation, negatives
+    assert KustoKqlCompiler._is_number_literal("5.") is False
+    assert KustoKqlCompiler._is_number_literal(".5") is False
+    assert KustoKqlCompiler._is_number_literal("-5") is False
+    assert KustoKqlCompiler._is_number_literal("-0.5") is False
+    assert KustoKqlCompiler._is_number_literal("1e10") is False
+    assert KustoKqlCompiler._is_number_literal("1.5e-3") is False
+
+    # Should NOT match: non-numeric strings
+    assert KustoKqlCompiler._is_number_literal("abc") is False
+    assert KustoKqlCompiler._is_number_literal("Measure 1") is False
+    assert KustoKqlCompiler._is_number_literal("") is False
+
+
+def test_calculated_measure_with_adhoc_measure_and_constant():
+    """Test calculated measure with an ad hoc measure and a constant.
+
+    Measure 1 = count(*), Measure 2 = "Measure 1" * 2
+    Measure 2 should compile to ["Measure 1"] * 2 (references the predefined measure)
+    """
+    measure_1 = literal_column("count(*)").label("Measure 1")
+    measure_2 = literal_column('"Measure 1" * 2').label("Measure 2")
+    query = select([measure_1, measure_2]).select_from(text("SalesData"))
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", "")
+    query_expected = (
+        '["SalesData"]'
+        '| summarize ["Measure 1"] = count() '
+        '| extend ["Measure 2"] = ["Measure 1"] * 2'
+        '| project ["Measure 1"], ["Measure 2"]'
+    )
+    assert query_compiled == query_expected
+
+
+def test_calculated_measure_with_two_adhoc_measures_and_aggregates():
+    """Test calculated measure referencing two ad hoc measures with aggregates.
+
+    Measure 1 = count(*), Measure 2 = count(*)
+    Measure 3 = "Measure 1" + "Measure 2" should compile to ["Measure 1"] + ["Measure 2"]
+    """
+    measure_1 = literal_column("count(*)").label("Measure 1")
+    measure_2 = literal_column("count(*)").label("Measure 2")
+    measure_3 = literal_column('"Measure 1" + "Measure 2"').label("Measure 3")
+    query = select([measure_1, measure_2, measure_3]).select_from(text("SalesData"))
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", "")
+    query_expected = (
+        '["SalesData"]'
+        '| summarize ["Measure 1"] = count(), ["Measure 2"] = count() '
+        '| extend ["Measure 3"] = ["Measure 1"] + ["Measure 2"]'
+        '| project ["Measure 1"], ["Measure 2"], ["Measure 3"]'
+    )
+    assert query_compiled == query_expected
+
+
+def test_calculated_measure_with_inline_aggregates():
+    """Test calculated measure with inline aggregates creating intermediary measures.
+
+    Measure = count("a") + count("b") should create intermediary measures:
+    - __Measure_1 = count(["a"])
+    - __Measure_2 = count(["b"])
+    - Measure = ["__Measure_1"] + ["__Measure_2"]
+    """
+    measure = literal_column('count("a") + count("b")').label("Measure")
+    query = select([measure]).select_from(text("SalesData"))
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", "")
+    query_expected = (
+        '["SalesData"]'
+        '| summarize ["__Measure_1"] = count(["a"]), ["__Measure_2"] = count(["b"]) '
+        '| extend ["Measure"] = ["__Measure_1"] + ["__Measure_2"]'
+        '| project ["Measure"]'
+    )
+    assert query_compiled == query_expected
+
+
+def test_calculated_measure_with_mixed_aggregates_and_references():
+    """Test calculated measure mixing inline aggregate and predefined measure.
+
+    Predefined 1 = count(*), Calculated = "Predefined 1" + count("b")
+    Should create:
+    - Predefined 1 = count()
+    - __Calculated_1 = count(["b"])
+    - Calculated = ["Predefined 1"] + ["__Calculated_1"]
+    """
+    predefined_1 = literal_column("count(*)").label("Predefined 1")
+    calculated = literal_column('"Predefined 1" + count("b")').label("Calculated")
+    query = select([predefined_1, calculated]).select_from(text("SalesData"))
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", "")
+    query_expected = (
+        '["SalesData"]'
+        '| summarize ["Predefined 1"] = count(), ["__Calculated_1"] = count(["b"]) '
+        '| extend ["Calculated"] = ["Predefined 1"] + ["__Calculated_1"]'
+        '| project ["Predefined 1"], ["Calculated"]'
+    )
+    assert query_compiled == query_expected
+
+
+# ============================================================================
+# Unit tests for helper functions
+# ============================================================================
+
+
+class TestFindTopLevelOperator:
+    """Tests for _find_top_level_operator helper."""
+
+    NOT_FOUND = -1
+    # Expected positions for operator in various test strings
+    POS_AFTER_SPACE_CHAR = 2  # "a + b" -> operator at index 2
+    POS_AFTER_PARENS = 4  # "(a) + (b)" -> operator at index 4
+    POS_AFTER_DOUBLE_PARENS = 6  # "((a)) + ((b))" -> operator at index 6
+    POS_AFTER_ESCAPED_QUOTE = 7  # '"a\"b" + c' -> operator at index 7
+    POS_MINUS = 6  # "a + b - c * d / e" -> minus at index 6
+    POS_MULT = 10  # "a + b - c * d / e" -> mult at index 10
+    POS_DIV = 14  # "a + b - c * d / e" -> div at index 14
+
+    def test_finds_operator_at_start(self):
+        assert KustoKqlCompiler._find_top_level_operator("+ b", "+") == 0
+
+    def test_finds_operator_in_middle(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator("a + b", "+")
+            == self.POS_AFTER_SPACE_CHAR
+        )
+
+    def test_finds_operator_at_end(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator("a +", "+")
+            == self.POS_AFTER_SPACE_CHAR
+        )
+
+    def test_not_found_returns_minus_one(self):
+        assert KustoKqlCompiler._find_top_level_operator("a b", "+") == self.NOT_FOUND
+
+    def test_operator_inside_double_quotes_not_found(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator('"a + b"', "+") == self.NOT_FOUND
+        )
+
+    def test_operator_inside_single_quotes_not_found(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator("'a + b'", "+") == self.NOT_FOUND
+        )
+
+    def test_operator_inside_brackets_not_found(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator('["a + b"]', "+")
+            == self.NOT_FOUND
+        )
+
+    def test_operator_inside_parens_not_found(self):
+        assert (
+            KustoKqlCompiler._find_top_level_operator("(a + b)", "+") == self.NOT_FOUND
+        )
+
+    def test_finds_operator_outside_parens(self):
+        result = KustoKqlCompiler._find_top_level_operator("(a) + (b)", "+")
+        assert result == self.POS_AFTER_PARENS
+
+    def test_finds_operator_with_nested_parens(self):
+        result = KustoKqlCompiler._find_top_level_operator("((a)) + ((b))", "+")
+        assert result == self.POS_AFTER_DOUBLE_PARENS
+
+    def test_mixed_quotes_and_operator(self):
+        result = KustoKqlCompiler._find_top_level_operator("\"col\" + 'value'", "+")
+        assert result == self.POS_AFTER_DOUBLE_PARENS
+
+    def test_operator_after_escaped_quote(self):
+        # Escaped quote should not affect detection
+        text = r'"a\"b" + c'
+        assert (
+            KustoKqlCompiler._find_top_level_operator(text, "+")
+            == self.POS_AFTER_ESCAPED_QUOTE
+        )
+
+    def test_multiple_operators_finds_first(self):
+        result = KustoKqlCompiler._find_top_level_operator("a + b + c", "+")
+        assert result == self.POS_AFTER_SPACE_CHAR
+
+    def test_different_operator_types(self):
+        text = "a + b - c * d / e"
+        assert (
+            KustoKqlCompiler._find_top_level_operator(text, "+")
+            == self.POS_AFTER_SPACE_CHAR
+        )
+        assert KustoKqlCompiler._find_top_level_operator(text, "-") == self.POS_MINUS
+        assert KustoKqlCompiler._find_top_level_operator(text, "*") == self.POS_MULT
+        assert KustoKqlCompiler._find_top_level_operator(text, "/") == self.POS_DIV
+
+    def test_empty_string(self):
+        assert KustoKqlCompiler._find_top_level_operator("", "+") == self.NOT_FOUND
+
+
+class TestCountOuterParens:
+    """Tests for _count_outer_parens helper."""
+
+    ZERO_PARENS = 0
+    ONE_PAREN = 1
+    TWO_PARENS = 2
+    THREE_PARENS = 3
+
+    def test_no_parens(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("a + b")
+        assert count == self.ZERO_PARENS
+        assert inner == "a + b"
+
+    def test_single_outer_paren(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("(a + b)")
+        assert count == self.ONE_PAREN
+        assert inner == "a + b"
+
+    def test_double_outer_parens(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("((a + b))")
+        assert count == self.TWO_PARENS
+        assert inner == "a + b"
+
+    def test_triple_outer_parens(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("(((x)))")
+        assert count == self.THREE_PARENS
+        assert inner == "x"
+
+    def test_parens_not_matching(self):
+        # (a) + (b) - first paren doesn't wrap the whole expression
+        count, inner = KustoKqlCompiler._count_outer_parens("(a) + (b)")
+        assert count == self.ZERO_PARENS
+        assert inner == "(a) + (b)"
+
+    def test_mixed_outer_and_inner(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("((a + (b)))")
+        assert count == self.TWO_PARENS
+        assert inner == "a + (b)"
+
+    def test_with_whitespace(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("  ( (x) )  ")
+        assert count == self.TWO_PARENS
+        assert inner == "x"
+
+    def test_empty_string(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("")
+        assert count == self.ZERO_PARENS
+        assert inner == ""
+
+    def test_single_char(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("x")
+        assert count == self.ZERO_PARENS
+        assert inner == "x"
+
+    def test_parens_only(self):
+        count, inner = KustoKqlCompiler._count_outer_parens("()")
+        assert count == self.ONE_PAREN
+        assert inner == ""
+
+
+class TestIsInsideQuotesOrBrackets:
+    """Tests for _is_inside_quotes_or_brackets helper."""
+
+    def test_position_inside_double_quotes(self):
+        text = 'before "inside" after'
+        # Positions inside the quotes (i, n, s, i, d, e)
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 8) is True
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 13) is True
+
+    def test_position_outside_double_quotes(self):
+        text = 'before "inside" after'
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 0) is False
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 5) is False
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 16) is False
+
+    def test_position_inside_single_quotes(self):
+        text = "before 'inside' after"
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 8) is True
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 13) is True
+
+    def test_position_outside_single_quotes(self):
+        text = "before 'inside' after"
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 0) is False
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 16) is False
+
+    def test_position_inside_brackets(self):
+        text = 'before ["column"] after'
+        # Position inside the brackets
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 8) is True
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 15) is True
+
+    def test_position_outside_brackets(self):
+        text = 'before ["column"] after'
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 0) is False
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 18) is False
+
+    def test_nested_quotes_in_brackets(self):
+        text = '["col with \\"quotes\\""]'
+        # Position inside should be True
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 5) is True
+
+    def test_position_at_boundary(self):
+        text = '"text"'
+        assert (
+            KustoKqlCompiler._is_inside_quotes_or_brackets(text, 0) is False
+        )  # At opening quote
+        assert (
+            KustoKqlCompiler._is_inside_quotes_or_brackets(text, 1) is True
+        )  # After opening quote
+
+    def test_out_of_bounds_position(self):
+        text = "short"
+        assert KustoKqlCompiler._is_inside_quotes_or_brackets(text, 100) is False
+
+
+class TestFindMatchingParen:
+    """Tests for _find_matching_paren helper."""
+
+    NOT_FOUND = -1
+    # Expected closing paren positions for various test strings
+    SIMPLE_CLOSE = 7  # "count(x)" -> closing paren at 7
+    OUTER_CLOSE = 12  # "sum(count(x))" -> outer closing at 12
+    INNER_CLOSE = 11  # "sum(count(x))" -> inner closing at 11
+    QUOTED_CLOSE = 23  # 'func("text(with)parens")' -> closing at 23
+    BRACKETED_CLOSE = 15  # 'func(["col(1)"])' -> closing at 15
+    EMPTY_CLOSE = 5  # "func()" -> closing at 5
+    DEEPLY_NESTED_OUTER = 9  # "a(b(c(d)))" -> outermost closing at 9
+    DEEPLY_NESTED_MID = 8  # "a(b(c(d)))" -> middle closing at 8
+    DEEPLY_NESTED_INNER = 7  # "a(b(c(d)))" -> innermost closing at 7
+
+    def test_simple_parentheses(self):
+        text = "count(x)"
+        assert KustoKqlCompiler._find_matching_paren(text, 5) == self.SIMPLE_CLOSE
+
+    def test_nested_parentheses(self):
+        text = "sum(count(x))"
+        assert KustoKqlCompiler._find_matching_paren(text, 3) == self.OUTER_CLOSE
+        assert KustoKqlCompiler._find_matching_paren(text, 9) == self.INNER_CLOSE
+
+    def test_parentheses_with_quotes(self):
+        # Parens inside quotes should be ignored
+        text = 'func("text(with)parens")'
+        assert KustoKqlCompiler._find_matching_paren(text, 4) == self.QUOTED_CLOSE
+
+    def test_parentheses_with_brackets(self):
+        # Parens inside brackets should be ignored
+        text = 'func(["col(1)"])'
+        assert KustoKqlCompiler._find_matching_paren(text, 4) == self.BRACKETED_CLOSE
+
+    def test_no_opening_paren_at_position(self):
+        text = "no paren here"
+        assert KustoKqlCompiler._find_matching_paren(text, 0) == self.NOT_FOUND
+
+    def test_unmatched_parenthesis(self):
+        text = "func(x"
+        assert KustoKqlCompiler._find_matching_paren(text, 4) == self.NOT_FOUND
+
+    def test_empty_parentheses(self):
+        text = "func()"
+        assert KustoKqlCompiler._find_matching_paren(text, 4) == self.EMPTY_CLOSE
+
+    def test_deeply_nested(self):
+        text = "a(b(c(d)))"
+        assert (
+            KustoKqlCompiler._find_matching_paren(text, 1) == self.DEEPLY_NESTED_OUTER
+        )
+        assert KustoKqlCompiler._find_matching_paren(text, 3) == self.DEEPLY_NESTED_MID
+        assert (
+            KustoKqlCompiler._find_matching_paren(text, 5) == self.DEEPLY_NESTED_INNER
+        )
+
+
+class TestHasOperatorsOutsideQuotes:
+    """Tests for _has_operators_outside_quotes helper."""
+
+    def test_simple_addition(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("a + b") is True
+
+    def test_simple_subtraction(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("a - b") is True
+
+    def test_simple_multiplication(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("a * b") is True
+
+    def test_simple_division(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("a / b") is True
+
+    def test_no_operators(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("count(x)") is False
+
+    def test_operator_inside_quotes(self):
+        # Note: _find_operator_outside_quotes only tracks double quotes, not single
+        assert KustoKqlCompiler._has_operators_outside_quotes('"a + b"') is False
+
+    def test_operator_inside_brackets(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes('["a+b"]') is False
+
+    def test_mixed_operators(self):
+        assert KustoKqlCompiler._has_operators_outside_quotes("a + b * c") is True
+
+    def test_aggregate_with_operators(self):
+        assert (
+            KustoKqlCompiler._has_operators_outside_quotes("count(a) + sum(b)") is True
+        )
+
+
+class TestExtractAndReplaceAggregates:
+    """Tests for _extract_and_replace_aggregates helper."""
+
+    ZERO_AGGS = 0
+    ONE_AGG = 1
+    TWO_AGGS = 2
+
+    def test_single_aggregate(self):
+        expr = 'count(["a"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        assert result == '["__Measure_1"]'
+        assert len(new_aggs) == self.ONE_AGG
+        assert new_aggs[0] == ('["__Measure_1"]', 'count(["a"])')
+
+    def test_two_aggregates_with_operator(self):
+        expr = 'count(["a"]) + sum(["b"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        assert result == '["__Measure_1"] + ["__Measure_2"]'
+        assert len(new_aggs) == self.TWO_AGGS
+        assert new_aggs[0][1] == 'count(["a"])'
+        assert new_aggs[1][1] == 'sum(["b"])'
+
+    def test_no_aggregates(self):
+        expr = '["a"] + ["b"]'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        assert result == '["a"] + ["b"]'
+        assert len(new_aggs) == self.ZERO_AGGS
+
+    def test_reuses_existing_aggregate(self):
+        expr = 'count(["a"]) + count(["a"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        # Both should use the same reference
+        assert result == '["__Measure_1"] + ["__Measure_1"]'
+        assert len(new_aggs) == self.ONE_AGG  # Only one unique aggregate
+
+    def test_existing_aggs_parameter(self):
+        existing = {'count(["a"])': '["existing_ref"]'}
+        expr = 'count(["a"]) + sum(["b"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure", existing
+        )
+        assert '["existing_ref"]' in result
+        assert len(new_aggs) == self.ONE_AGG  # Only sum is new, count is reused
+
+    def test_aggregate_in_quotes_ignored(self):
+        expr = '"count(a)"'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        assert result == '"count(a)"'
+        assert len(new_aggs) == self.ZERO_AGGS
+
+    def test_aggregate_in_brackets_ignored(self):
+        expr = '["count(a)"]'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, "Measure"
+        )
+        assert result == '["count(a)"]'
+        assert len(new_aggs) == self.ZERO_AGGS
+
+    def test_measure_name_with_special_chars(self):
+        expr = 'count(["a"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(
+            expr, '["My Measure"]'
+        )
+        assert result == '["__My Measure_1"]'
+
+    def test_complex_expression(self):
+        expr = 'count(["a"]) * 100 / sum(["b"])'
+        result, new_aggs = KustoKqlCompiler._extract_and_replace_aggregates(expr, "Pct")
+        assert '["__Pct_1"]' in result
+        assert '["__Pct_2"]' in result
+        assert "* 100 /" in result
+        assert len(new_aggs) == self.TWO_AGGS
+
+
+class TestContainsAggregateFunction:
+    """Tests for _contains_aggregate_function helper."""
+
+    def test_contains_count(self):
+        assert KustoKqlCompiler._contains_aggregate_function("count(x)") is True
+
+    def test_contains_sum(self):
+        assert KustoKqlCompiler._contains_aggregate_function("sum(x)") is True
+
+    def test_contains_avg(self):
+        assert KustoKqlCompiler._contains_aggregate_function("avg(x)") is True
+
+    def test_contains_min_max(self):
+        assert KustoKqlCompiler._contains_aggregate_function("min(x)") is True
+        assert KustoKqlCompiler._contains_aggregate_function("max(x)") is True
+
+    def test_contains_dcount(self):
+        assert KustoKqlCompiler._contains_aggregate_function("dcount(x)") is True
+
+    def test_no_aggregate(self):
+        assert KustoKqlCompiler._contains_aggregate_function("col + 1") is False
+        assert KustoKqlCompiler._contains_aggregate_function('["column"]') is False
+
+    def test_aggregate_in_quotes_not_counted(self):
+        assert KustoKqlCompiler._contains_aggregate_function('"count(x)"') is False
+        assert KustoKqlCompiler._contains_aggregate_function("'sum(x)'") is False
+
+    def test_aggregate_in_brackets_not_counted(self):
+        assert KustoKqlCompiler._contains_aggregate_function('["count(x)"]') is False
+
+    def test_aggregate_with_complex_arg(self):
+        assert (
+            KustoKqlCompiler._contains_aggregate_function('count(["My Col"])') is True
+        )
+
+    def test_multiple_aggregates(self):
+        assert (
+            KustoKqlCompiler._contains_aggregate_function("count(a) + sum(b)") is True
+        )
+
+
 @pytest.mark.parametrize(
     ("query_table_name", "expected_table_name"),
     [