refactor: set cascade_backrefs=False for SqlaTable

pytest.ini

@@ -31,7 +31,7 @@ filterwarnings =
     error:Passing a string to Connection.execute\(\) is deprecated:sqlalchemy.exc.RemovedIn20Warning
 #    error:"Query" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
 #    error:"SavedQuery" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-#    error:"SqlaTable" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
+    error:"SqlaTable" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
 #    error:"SqlMetric" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
 #    error:"TableColumn" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
 #    error:"TaggedObject" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning

superset/connectors/sqla/models.py

@@ -1364,7 +1364,14 @@ class SqlaTable(
 
     database: Database = relationship(
         "Database",
-        backref=backref("tables", cascade="all, delete-orphan"),
+        backref=backref(
+            "tables",
+            cascade="all, delete-orphan",
+            # SQLAlchemy 2.0 behavior: assigning `table.database` no longer
+            # cascades the SqlaTable into the Database's session; callers must
+            # add objects to a session explicitly.
+            cascade_backrefs=False,
+        ),
         foreign_keys=[database_id],
     )
     schema = Column(String(255))

superset/examples/generic_loader.py

@@ -173,6 +173,13 @@ def safe_serialize(x: Any, column_name: str) -> Optional[str]:
 
     if not tbl:
         tbl = SqlaTable(table_name=table_name, database_id=database.id)
+        # Explicitly add the new table to the session. Assigning `tbl.database`
+        # below no longer implicitly adds `tbl` to the session (SQLAlchemy 2.0
+        # behavior, cascade_backrefs=False), so without this, the two
+        # `db.session.merge()` calls below (one inside `fetch_metadata()`, one
+        # at the end of this function) would each create a separate transient
+        # copy of `tbl`, resulting in two pending inserts for the same uuid.
+        db.session.add(tbl)
         # Set the database reference
         tbl.database = database
 
@@ -187,7 +194,7 @@ def safe_serialize(x: Any, column_name: str) -> Optional[str]:
         tbl.fetch_metadata()
 
     db.session.merge(tbl)
-    db.session.commit()
+    db.session.commit()  # pylint: disable=consider-using-transaction
 
     return tbl
 
@@ -242,7 +249,7 @@ def loader(
         if description and tbl:
             tbl.description = description
             db.session.merge(tbl)
-            db.session.commit()
+            db.session.commit()  # pylint: disable=consider-using-transaction
 
     # Set function name and docstring
     loader.__name__ = f"load_{parquet_file}"

tests/integration_tests/sqla_models_tests.py

@@ -173,6 +173,7 @@ def test_jinja_metrics_and_calc_columns(self, mock_username: MagicMock) -> None:
             "'{{ 'xyz_' + time_grain }}' as time_grain",
             database=get_example_database(),
         )
+        db.session.add(table)
         TableColumn(
             column_name="expr",
             expression="case when '{{ current_username() }}' = 'abc' "
@@ -275,6 +276,7 @@ def test_adhoc_metrics_and_calc_columns(self):
         table = SqlaTable(
             table_name="test_validate_adhoc_sql", database=get_example_database()
         )
+        db.session.add(table)
         db.session.commit()
 
         with pytest.raises(QueryObjectValidationError):