Feat: allow tables to be preserved in replace_table (#4468)

* Feat: allow tables to be preserved in replace_table

* PR feedback

* PR feedback

Author: georgesittas

sqlglot/dialects/dialect.py

@@ -8,7 +8,7 @@
 from sqlglot import exp
 from sqlglot.errors import ParseError
 from sqlglot.generator import Generator, unsupported_args
-from sqlglot.helper import AutoName, flatten, is_int, seq_get, subclasses
+from sqlglot.helper import AutoName, flatten, is_int, seq_get, subclasses, to_bool
 from sqlglot.jsonpath import JSONPathTokenizer, parse as parse_json_path
 from sqlglot.parser import Parser
 from sqlglot.time import TIMEZONES, format_time, subsecond_precision
@@ -770,14 +770,7 @@ def get_or_raise(cls, dialect: DialectType) -> Dialect:
                     elif len(pair) == 2:
                         value = pair[1].strip()
 
-                        # Coerce the value to boolean if it matches to the truthy/falsy values below
-                        value_lower = value.lower()
-                        if value_lower in ("true", "1"):
-                            value = True
-                        elif value_lower in ("false", "0"):
-                            value = False
-
-                    kwargs[key] = value
+                    kwargs[key] = to_bool(value)
 
             except ValueError:
                 raise ValueError(

sqlglot/expressions.py

@@ -31,6 +31,7 @@
     ensure_list,
     seq_get,
     subclasses,
+    to_bool,
 )
 from sqlglot.tokens import Token, TokenError
 
@@ -312,7 +313,7 @@ def add_comments(self, comments: t.Optional[t.List[str]] = None, prepend: bool =
                     for kv in "".join(meta).split(","):
                         k, *v = kv.split("=")
                         value = v[0].strip() if v else True
-                        self.meta[k.strip()] = value
+                        self.meta[k.strip()] = to_bool(value)
 
                 if not prepend:
                     self.comments.append(comment)
@@ -8231,7 +8232,7 @@ def replace_tables(
     mapping = {normalize_table_name(k, dialect=dialect): v for k, v in mapping.items()}
 
     def _replace_tables(node: Expression) -> Expression:
-        if isinstance(node, Table):
+        if isinstance(node, Table) and node.meta.get("replace") is not False:
             original = normalize_table_name(node, dialect=dialect)
             new_name = mapping.get(original)
 

sqlglot/helper.py

@@ -456,6 +456,20 @@ def first(it: t.Iterable[T]) -> T:
     return next(i for i in it)
 
 
+def to_bool(value: t.Optional[str | bool]) -> t.Optional[str | bool]:
+    if isinstance(value, bool) or value is None:
+        return value
+
+    # Coerce the value to boolean if it matches to the truthy/falsy values below
+    value_lower = value.lower()
+    if value_lower in ("true", "1"):
+        return True
+    if value_lower in ("false", "0"):
+        return False
+
+    return value
+
+
 def merge_ranges(ranges: t.List[t.Tuple[A, A]]) -> t.List[t.Tuple[A, A]]:
     """
     Merges a sequence of ranges, represented as tuples (low, high) whose values

tests/test_expressions.py

@@ -258,6 +258,14 @@ def test_replace_tables(self):
             'SELECT * FROM "my-project"."example"."table" /* example.table */',
         )
 
+        self.assertEqual(
+            exp.replace_tables(
+                parse_one("select * from example.table /* sqlglot.meta replace=false */"),
+                {"example.table": "a.b"},
+            ).sql(),
+            "SELECT * FROM example.table /* sqlglot.meta replace=false */",
+        )
+
     def test_expand(self):
         self.assertEqual(
             exp.expand(
@@ -1168,7 +1176,7 @@ def test_is_type(self):
 
     def test_set_meta(self):
         query = parse_one("SELECT * FROM foo /* sqlglot.meta x = 1, y = a, z */")
-        self.assertEqual(query.find(exp.Table).meta, {"x": "1", "y": "a", "z": True})
+        self.assertEqual(query.find(exp.Table).meta, {"x": True, "y": "a", "z": True})
         self.assertEqual(query.sql(), "SELECT * FROM foo /* sqlglot.meta x = 1, y = a, z */")
 
     def test_assert_is(self):