bump: dbt Core to 1.12

.flake8

@@ -3,10 +3,11 @@ select =
     E
     W
     F
+# W503 and E203 make Flake8 work like Black.
 ignore =
-    W503 # makes Flake8 work like black
+    W503
     W504
-    E203 # makes Flake8 work like black
+    E203
     E741
     E501
 exclude = tests

.github/workflows/ci.yml

@@ -6,6 +6,26 @@ on:
   workflow_dispatch:
 
 jobs:
+  unit-tests:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.10', '3.13']
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install unit test dependencies
+      run: python3 -m pip install . dbt-tests-adapter==1.20.0 pytest pytest-dotenv
+    - name: Check installed dependencies
+      run: python3 -m pip check
+    - name: Run unit tests
+      run: python3 -m pytest tests/unit
+
   test-with-jaffle_shop:
     runs-on: ubuntu-latest
     strategy:
@@ -41,11 +61,11 @@ jobs:
         EOF
     - name: Install dbt-risingwave globally
       run: python3 -m pip install .
-      
+
     - name: dbt-seed
       run: dbt seed
       working-directory: jaffle_shop
-      
+
     - name: dbt-run
       run: dbt run
       working-directory: jaffle_shop

dbt/adapters/risingwave/__version__.py

@@ -1 +1 @@
-version = "1.11.16"
+version = "1.12.0"

dbt/adapters/risingwave/connections.py

@@ -8,6 +8,8 @@
     PostgresConnectionManager,
     PostgresCredentials,
 )
+from dbt.adapters.postgres.record import PostgresRecordReplayHandle
+from dbt_common.record import RecorderMode, get_record_mode_from_env
 
 logger = AdapterLogger("RisingWave")
 
@@ -71,6 +73,7 @@ def _connection_keys(self):
             "sslmode",
             "keepalives_idle",
             "connect_timeout",
+            "autocommit",
             "retries",
         )
 
@@ -119,15 +122,29 @@ def _super_open(cls, connection, extra_kwargs: Optional[Dict[str, str]] = None):
         kwargs.update(extra_kwargs or {})
 
         def connect():
-            handle = psycopg2.connect(
-                dbname=credentials.database,
-                user=credentials.user,
-                host=credentials.host,
-                password=credentials.password,
-                port=credentials.port,
-                connect_timeout=credentials.connect_timeout,
-                **kwargs,
-            )
+            handle = None
+
+            # Keep this in sync with PostgresConnectionManager.open. Replay
+            # mode does not create a real database connection, while record
+            # and diff modes wrap one to observe native connection activity.
+            rec_mode = get_record_mode_from_env()
+            if rec_mode != RecorderMode.REPLAY:
+                handle = psycopg2.connect(
+                    dbname=credentials.database,
+                    user=credentials.user,
+                    host=credentials.host,
+                    password=credentials.password,
+                    port=credentials.port,
+                    connect_timeout=credentials.connect_timeout,
+                    **kwargs,
+                )
+
+            if handle is not None and credentials.autocommit:
+                handle.autocommit = True
+
+            if rec_mode is not None:
+                handle = PostgresRecordReplayHandle(handle, connection)
+
             if credentials.role:
                 handle.cursor().execute("set role {}".format(credentials.role))
             return handle

dev-requirements.txt

@@ -1,21 +1,22 @@
-# install latest changes in dbt-core
-git+https://github.com/dbt-labs/dbt-core.git@main#subdirectory=core
-git+https://github.com/dbt-labs/dbt-adapters.git@main#subdirectory=dbt-tests-adapter
-git+https://github.com/dbt-labs/dbt-common.git@main
-git+https://github.com/dbt-labs/dbt-postgres.git@main
+# Test against the released dbt versions supported by setup.py. Pinning these
+# keeps local and tox environments from accidentally installing dbt 2.0 work
+# from upstream main branches.
+dbt-core==1.12.2
+dbt-tests-adapter==1.20.0
+dbt-common==1.39.0
+dbt-postgres==1.11.0
 
-black==22.3.0
+black==24.2.0
 bumpversion
 flake8
 flaky
 freezegun==0.3.12
 ipdb
-mypy==0.782
+mypy==1.8.0
 pip-tools
 pre-commit
 pytest
 pytest-dotenv
-pytest-logbook
 pytest-csv
 pytest-xdist
 pytz

setup.py

@@ -32,24 +32,21 @@ def _plugin_version() -> str:
     packages=find_namespace_packages(include=["dbt", "dbt.*"]),
     include_package_data=True,
     install_requires=[
-        "dbt-postgres~=1.9.0",
-        "dbt-core~=1.11.0",
-        # not sure if these are needed due to inheritance from dbt-postgres
-        # but doesn't hurt to be explicit I suppose
-        "dbt-common>=1.0.4,<2.0",
-        "dbt-adapters>=1.7.0,<2.0",
+        "dbt-postgres~=1.11.0",
+        "dbt-core~=1.12.0",
+        "dbt-common>=1.37.5,<2.0",
+        "dbt-adapters>=1.24.5,<2.0",
     ],
     classifiers=[
         "Development Status :: 5 - Production/Stable",
         "License :: OSI Approved :: Apache Software License",
         "Operating System :: Microsoft :: Windows",
         "Operating System :: MacOS :: MacOS X",
         "Operating System :: POSIX :: Linux",
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Programming Language :: Python :: 3.12",
         "Programming Language :: Python :: 3.13",
     ],
-    python_requires=">=3.9",
+    python_requires=">=3.10",
 )

tests/unit/test_connections.py

@@ -2,7 +2,7 @@
 import sys
 from pathlib import Path
 from types import SimpleNamespace
-from unittest.mock import Mock, call
+from unittest.mock import Mock, call, patch
 
 
 CONNECTIONS = (
@@ -14,6 +14,100 @@
 )
 
 
+def test_open_passes_risingwave_options_and_enables_autocommit():
+    connections = load_local_connections_module()
+    credentials = connections.RisingWaveCredentials.from_dict(
+        {
+            "host": "127.0.0.1",
+            "user": "root",
+            "password": "",
+            "port": 4566,
+            "dbname": "dev",
+            "schema": "public",
+            "autocommit": True,
+        }
+    )
+    connection = SimpleNamespace(state="init", credentials=credentials, handle=None)
+    handle = SimpleNamespace(autocommit=False)
+
+    def retry_connection(connection, connect, **kwargs):
+        connection.handle = connect()
+        connection.state = "open"
+        return connection
+
+    with (
+        patch.object(connections, "get_record_mode_from_env", return_value=None),
+        patch.object(connections.psycopg2, "connect", return_value=handle) as connect,
+        patch.object(
+            connections.RisingWaveConnectionManager,
+            "retry_connection",
+            side_effect=retry_connection,
+        ),
+    ):
+        result = connections.RisingWaveConnectionManager._super_open(
+            connection, extra_kwargs={"gssencmode": "disable"}
+        )
+
+    assert result is connection
+    assert handle.autocommit is True
+    assert "autocommit" in credentials._connection_keys()
+    connect.assert_called_once_with(
+        dbname="dev",
+        user="root",
+        host="127.0.0.1",
+        password="",
+        port=4566,
+        connect_timeout=10,
+        application_name="dbt",
+        gssencmode="disable",
+    )
+
+
+def test_open_uses_record_replay_handle_without_real_connection():
+    connections = load_local_connections_module()
+    credentials = connections.RisingWaveCredentials.from_dict(
+        {
+            "host": "127.0.0.1",
+            "user": "root",
+            "password": "",
+            "port": 4566,
+            "dbname": "dev",
+            "schema": "public",
+        }
+    )
+    connection = SimpleNamespace(state="init", credentials=credentials, handle=None)
+    replay_handle = object()
+
+    def retry_connection(connection, connect, **kwargs):
+        connection.handle = connect()
+        connection.state = "open"
+        return connection
+
+    with (
+        patch.object(
+            connections,
+            "get_record_mode_from_env",
+            return_value=connections.RecorderMode.REPLAY,
+        ),
+        patch.object(connections.psycopg2, "connect") as connect,
+        patch.object(
+            connections,
+            "PostgresRecordReplayHandle",
+            return_value=replay_handle,
+        ) as record_replay_handle,
+        patch.object(
+            connections.RisingWaveConnectionManager,
+            "retry_connection",
+            side_effect=retry_connection,
+        ),
+    ):
+        result = connections.RisingWaveConnectionManager._super_open(connection)
+
+    assert result.handle is replay_handle
+    connect.assert_not_called()
+    record_replay_handle.assert_called_once_with(None, connection)
+
+
 def test_cancel_quotes_compound_process_id_with_query_binding():
     connections = load_local_connections_module()
     manager = connections.RisingWaveConnectionManager.__new__(

tox.ini

@@ -1,8 +1,8 @@
 [tox]
 skipsdist = True
-envlist = py37,py38,py39,py310,py311
+envlist = py310,py311,py312,py313
 
-[testenv:{unit,py37,py38,py39,py310,py311,py}]
+[testenv:{unit,py310,py311,py312,py313,py}]
 description = unit testing
 skip_install = true
 passenv =
@@ -13,7 +13,7 @@ deps =
   -rdev-requirements.txt
   -e.
 
-[testenv:{integration,py37,py38,py39,py310,py311,py}-{ risingwave }]
+[testenv:{integration,py310,py311,py312,py313,py}-{ risingwave }]
 description = adapter plugin integration testing
 skip_install = true
 passenv =