diff --git a/sqlalchemy_kusto/dialect_kql.py b/sqlalchemy_kusto/dialect_kql.py
index 5be6be3..508f3f9 100644
--- a/sqlalchemy_kusto/dialect_kql.py
+++ b/sqlalchemy_kusto/dialect_kql.py
@@ -3,7 +3,6 @@
 
 from sqlalchemy import Column, exc, sql
 from sqlalchemy.sql import compiler, operators, selectable
-from sqlalchemy.sql.compiler import OPERATORS
 
 from sqlalchemy_kusto.dialect_base import KustoBaseDialect
 
@@ -82,7 +81,6 @@ def __init__(self, dialect, **kw):
 
 
 class KustoKqlCompiler(compiler.SQLCompiler):
-    OPERATORS[operators.and_] = " and "
     delete_extra_from_clause = None
     update_from_clause = None
     visit_empty_set_expr = None
@@ -162,6 +160,18 @@ def visit_select(
     def limit_clause(self, select, **kw):
         return ""
 
+    def visit_clauselist(self, clauselist, **kw):
+        kql_operators = {
+            operators.and_: " and ",
+            operators.or_: " or ",
+        }
+        
+        if clauselist.operator in kql_operators:
+            sep = kql_operators[clauselist.operator]
+            return self._generate_delimited_list(clauselist.clauses, sep, **kw)
+            
+        return super().visit_clauselist(clauselist, **kw)
+
     def _legacy_join(self, select_stmt: selectable.Select, **kwargs):
         """Consumes arguments from join() or outerjoin(), places them into a
         consistent format with which to form the actual JOIN constructs.
@@ -528,11 +538,6 @@ def _sql_to_kql_where(where_clause: str) -> str:
             where_clause,
             flags=re.IGNORECASE,
         )
-        # Handle logical operators 'AND' and 'OR' to ensure the conditions are preserved
-        # Replace AND with 'and' in KQL
-        where_clause = re.sub(r"\s+AND\s+", r" and ", where_clause, flags=re.IGNORECASE)
-        # Replace OR with 'or' in KQL
-        where_clause = re.sub(r"\s+OR\s+", r" or ", where_clause, flags=re.IGNORECASE)
         return where_clause
 
     @staticmethod
diff --git a/tests/unit/test_dialect_kql.py b/tests/unit/test_dialect_kql.py
index 04b8fc0..40ec817 100644
--- a/tests/unit/test_dialect_kql.py
+++ b/tests/unit/test_dialect_kql.py
@@ -576,3 +576,51 @@ def test_schema_from_query(query_table_name: str, expected_table_name: str):
 
     query_expected = f"let inner_qry = ({expected_table_name});inner_qry| take 5"
     assert query_compiled == query_expected
+
+
+def test_logical_operators_and_literals_precedence():
+    val_and = "DATA AND ANALYTICS"
+    val_or = "OPEN OR CLOSED"
+
+    condition = sa.and_(
+        sa.or_(column("Field1") == val_and, column("Field2") == val_or),
+        column("Status") == "ACTIVE",
+    )
+
+    query = select([column("Field1")]).select_from(text("logs")).where(condition)
+
+    query_compiled = str(
+        query.compile(engine, compile_kwargs={"literal_binds": True})
+    ).replace("\n", " ")
+
+    expected_full = (
+        '["logs"] '
+        f"| where ([\"Field1\"] == '{val_and}' or [\"Field2\"] == '{val_or}') and [\"Status\"] == 'ACTIVE' "
+        '| project ["Field1"]'
+    )
+    assert query_compiled == expected_full
+
+
+def test_logical_operators_precedence_and_casing():
+    """
+    Test that 'or' inside 'and' gets parentheses (required),
+    but 'and' inside 'or' does not (not required by precedence).
+    """
+
+    cond_nested_or = sa.and_(
+        sa.or_(column("A") == 1, column("B") == 2),
+        column("C") == 3
+    )
+    query1 = select([column("Field")]).select_from(text("logs")).where(cond_nested_or)
+    compiled1 = str(query1.compile(engine, compile_kwargs={"literal_binds": True})).replace("\n", " ")
+
+    assert '(["A"] == 1 or ["B"] == 2) and ["C"] == 3' in compiled1
+
+    cond_nested_and = sa.or_(
+        column("A") == 1,
+        sa.and_(column("B") == 2, column("C") == 3)
+    )
+    query2 = select([column("Field")]).select_from(text("logs")).where(cond_nested_and)
+    compiled2 = str(query2.compile(engine, compile_kwargs={"literal_binds": True})).replace("\n", " ")
+
+    assert '["A"] == 1 or ["B"] == 2 and ["C"] == 3' in compiled2