dbt debug flow: что должно работать минимально

dbt debug — это первый тест, который должен пройти любой adapter. Если debug passes — у вас есть рабочее соединение с warehouse, и можно начинать тестировать. Если падает — что-то фундаментально сломано.

В этом уроке разбираем, что именно dbt debug делает, какие методы вашего adapter’а он использует, и как minimal реализация выглядит для passing debug.

Что делает dbt debug

dbt debug выполняет несколько проверок, чтобы убедиться что dbt может работать с вашим warehouse:

Если все 8 шагов pass — adapter готов для basic use. Иначе debug говорит, какой step failed.

Output successful debug

$ dbt debug --profiles-dir . --profile dbt-oceanbase

Running with dbt=1.8.5
dbt version: 1.8.5
python version: 3.11.6
python path: /opt/venv/bin/python
os info: Linux-5.15.0-x86_64

Profile setup:
  profiles.yml file [OK found and valid]
  dbt_project.yml file [OK found and valid]

Profile: dbt-oceanbase

Configuration:
  Required dependencies:
   - git [OK found]

Connection:
  host: localhost
  port: 2881
  user: root
  database: test
  schema: public
  Connection test: [OK connection ok]
  Required adapter capabilities: [OK all supported]

All checks passed!

Если что-то падает — debug пишет error на том месте. Easy для localization.

Что minimal нужно для passing debug

Чтобы dbt debug passes, ваш adapter должен иметь:

1. Properly зарегистрированный AdapterPlugin

# dbt/adapters/oceanbase/__init__.py
from dbt.adapters.oceanbase.connections import OceanBaseConnectionManager
from dbt.adapters.oceanbase.credentials import OceanBaseCredentials
from dbt.adapters.oceanbase.impl import OceanBaseAdapter
from dbt.adapters.base import AdapterPlugin
from dbt.include import oceanbase

Plugin = AdapterPlugin(
    adapter=OceanBaseAdapter,
    credentials=OceanBaseCredentials,
    include_path=oceanbase.PACKAGE_PATH,
)

И в dbt/include/oceanbase/__init__.py:

from pathlib import Path
PACKAGE_PATH = Path(__file__).parent

Без этого dbt не знает что adapter существует.

2. Credentials с required fields

Required fields в Credentials должны соответствовать profiles.yml:

@dataclass
class OceanBaseCredentials(Credentials):
    host: str       # ← required
    user: str       # ← required
    password: str   # ← required
    database: str   # ← inherited from Credentials base
    schema: str     # ← inherited
    
    port: int = 2881  # ← optional с default
    
    @property
    def type(self) -> str:
        return 'oceanbase'

Если profile_dir profiles.yml не имеет required field — debug fails immediately.

3. ConnectionManager с working open()

Минимум для open():

@classmethod
def open(cls, connection):
    if connection.state == ConnectionState.OPEN:
        return connection
    
    credentials = connection.credentials
    handle = pymysql.connect(
        host=credentials.host,
        port=credentials.port,
        user=credentials.user,
        password=credentials.password,
        database=credentials.database,
    )
    
    connection.handle = handle
    connection.state = ConnectionState.OPEN
    return connection

Должно вернуть Connection с handle != None.

4. Working execute() через cursor

Для SELECT 1 test query — нужен work через SQLAdapter.execute():

class OceanBaseAdapter(SQLAdapter):
    ConnectionManager = OceanBaseConnectionManager
    
    # execute() inherited from SQLAdapter
    # Uses self.add_query() -> connection.handle.cursor().execute(sql)

Если ваш connection.handle имеет standard cursor API — это inherited code работает.

5. list_schemas macro

Для check 6 (list schemas) — нужен:

-- dbt/include/oceanbase/macros/adapters.sql
{% macro oceanbase__list_schemas(database) %}
  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
    SHOW DATABASES
  {% endcall %}
  {{ return(load_result('list_schemas').table) }}
{% endmacro %}

Или default version из SQLAdapter работает для ANSI SQL warehouses (information_schema.schemata).

6. check_schema_exists macro

{% macro oceanbase__check_schema_exists(information_schema, schema) %}
  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) %}
    SELECT COUNT(*) FROM information_schema.schemata
    WHERE schema_name = '{{ schema }}'
  {% endcall %}
  {{ return(load_result('check_schema_exists').table) }}
{% endmacro %}

Минимум кода для passing debug

Полный minimal adapter для passing debug (без работы dbt run):

# dbt/adapters/myadapter/credentials.py
from dataclasses import dataclass
from dbt.adapters.contracts.connection import Credentials

@dataclass
class MyAdapterCredentials(Credentials):
    host: str
    user: str = ''
    password: str = ''
    database: str = ''
    schema: str = ''
    port: int = 5432
    
    @property
    def type(self):
        return 'myadapter'
    
    @property
    def unique_field(self):
        return self.host
    
    def _connection_keys(self):
        return ('host', 'port', 'user', 'database', 'schema')

# dbt/adapters/myadapter/connections.py
from contextlib import contextmanager
import sqlite3

from dbt.adapters.base import BaseConnectionManager
from dbt.adapters.contracts.connection import (
    AdapterResponse, Connection, ConnectionState,
)


class MyAdapterConnectionManager(BaseConnectionManager):
    TYPE = 'myadapter'

    @classmethod
    def open(cls, connection):
        if connection.state == ConnectionState.OPEN:
            return connection
        
        # Use SQLite for demo
        handle = sqlite3.connect(':memory:')
        connection.handle = handle
        connection.state = ConnectionState.OPEN
        return connection

    @classmethod
    def get_response(cls, cursor):
        return AdapterResponse(_message='OK', rows_affected=cursor.rowcount)

    @contextmanager
    def exception_handler(self, sql):
        try:
            yield
        except sqlite3.Error as e:
            self.release()
            raise RuntimeError(str(e))

    def cancel(self, connection):
        connection.handle.close()

# dbt/adapters/myadapter/impl.py
from dbt.adapters.sql import SQLAdapter
from dbt.adapters.myadapter.connections import MyAdapterConnectionManager


class MyAdapterAdapter(SQLAdapter):
    ConnectionManager = MyAdapterConnectionManager
    # Relation = ... (use default for now)
    # Column = ... (use default)
    
    @classmethod
    def date_function(cls):
        return 'DATE(\\'now\\')'  # SQLite syntax
    
    @classmethod
    def convert_text_type(cls, agate_table, col_idx):
        return 'TEXT'
    
    @classmethod
    def convert_number_type(cls, agate_table, col_idx):
        return 'NUMERIC'
    
    @classmethod
    def convert_boolean_type(cls, agate_table, col_idx):
        return 'BOOLEAN'
    
    @classmethod
    def convert_datetime_type(cls, agate_table, col_idx):
        return 'DATETIME'
    
    @classmethod
    def convert_date_type(cls, agate_table, col_idx):
        return 'DATE'
    
    @classmethod
    def convert_time_type(cls, agate_table, col_idx):
        return 'TIME'

# dbt/adapters/myadapter/__init__.py
from dbt.adapters.myadapter.connections import MyAdapterConnectionManager
from dbt.adapters.myadapter.credentials import MyAdapterCredentials
from dbt.adapters.myadapter.impl import MyAdapterAdapter
from dbt.adapters.base import AdapterPlugin
from dbt.include import myadapter

Plugin = AdapterPlugin(
    adapter=MyAdapterAdapter,
    credentials=MyAdapterCredentials,
    include_path=myadapter.PACKAGE_PATH,
)

# dbt/include/myadapter/__init__.py
from pathlib import Path
PACKAGE_PATH = Path(__file__).parent

-- dbt/include/myadapter/macros/adapters.sql
{% macro myadapter__list_schemas(database) %}
  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
    SELECT name AS schema_name FROM sqlite_master WHERE type='table'
  {% endcall %}
  {{ return(load_result('list_schemas').table) }}
{% endmacro %}

{% macro myadapter__check_schema_exists(information_schema, schema) %}
  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) %}
    SELECT 1 AS schema_exists
  {% endcall %}
  {{ return(load_result('check_schema_exists').table) }}
{% endmacro %}

{% macro myadapter__create_schema(relation) %}
  {# SQLite не имеет CREATE SCHEMA, noop #}
{% endmacro %}

{% macro myadapter__drop_schema(relation) %}
  {# SQLite не имеет DROP SCHEMA, noop #}
{% endmacro %}

# dbt/include/myadapter/dbt_project.yml
config-version: 2
name: myadapter
version: '1.0'

profile: 'default'

target-path: 'target'
clean-targets: ['target', 'dbt_packages']

models:
  myadapter:
    +materialized: view

# setup.py
from setuptools import setup, find_namespace_packages

setup(
    name='dbt-myadapter',
    version='1.0.0',
    packages=find_namespace_packages(include=['dbt.*']),
    package_data={
        'dbt': [
            'include/myadapter/dbt_project.yml',
            'include/myadapter/macros/*.sql',
        ],
    },
    install_requires=[
        'dbt-core>=1.8',
        'dbt-adapters>=1.7',
    ],
)

Install:

pip install -e .

Profile:

# ~/.dbt/profiles.yml
test-myadapter:
  target: dev
  outputs:
    dev:
      type: myadapter
      host: localhost
      user: test
      database: test
      schema: main

Run:

dbt debug --profile test-myadapter
# Should print: 'All checks passed!'

Common debug failures и fixes

Failure 1: Adapter not found

Encountered an error:
  Runtime Error
    Could not find adapter type 'myadapter'!

Cause: AdapterPlugin не зарегистрирован.

Debug:

python -c "from dbt.adapters.factory import FACTORY; print(FACTORY.plugins.keys())"

Если ‘myadapter’ не в keys — Plugin не registered.

Fix: ensure pip install -e . succeeded. Check dbt/adapters/myadapter/__init__.py имеет Plugin = AdapterPlugin(...).

Failure 2: Missing required field

Encountered an error:
  Compilation Error
    Credentials in profile "test-myadapter", target "dev" invalid:
    'host' is a required property

Cause: profiles.yml не имеет required field defined в Credentials dataclass.

Fix: add host: к profiles.yml или make optional в dataclass.

Failure 3: Connection failed

Encountered an error:
  Database Error
    Could not connect to database:
    [Errno 111] Connection refused

Cause: warehouse не запущен на host/port.

Debug:

nc -zv localhost 5432   # Postgres
nc -zv localhost 2881   # OceanBase

If connection refused — warehouse не slušа на этом port.

Fix: start warehouse, verify port, check firewall.

Failure 4: Authentication failed

Encountered an error:
  Database Error
    Authentication failed for user 'wrong_user'

Cause: wrong credentials.

Fix: verify user/password/auth method.

Failure 5: list_schemas macro fails

Encountered an error:
  Database Error in query 'list_schemas'
    no such function: information_schema.schemata

Cause: default list_schemas использует ANSI SQL, не работает на SQLite/non-standard warehouses.

Fix: override через <adapter>__list_schemas macro.

debug exit codes

dbt debug; echo $?

• 0: all checks passed
• 2: some checks failed
• 1: unknown error

Use в CI:

- name: Verify dbt can connect
  run: dbt debug

CI fails if debug doesn’t pass — early detection of config issues.

Попробуй сам

1. Создайте minimal adapter scaffold (как выше).

2. pip install -e . для регистрации.

3. Create profiles.yml.

4. Run dbt debug. Should pass.

5. Намеренно сломайте:

   • Remove host from profiles.yml — должно fail на step 2
   • Wrong port в profiles.yml — должно fail на step 4 (connection)
   • Remove oceanbase__list_schemas macro — должно fail на step 6 (if no fallback)

6. Add to CI:

   - run: dbt debug

   Verify CI fails when config broken.

Ключевые выводы

1. dbt debug — 8 checks: YAML syntax -> Credentials -> Plugin lookup -> open() -> SELECT 1 -> list_schemas -> check_schema_exists -> summary.

2. Minimum для passing debug: registered AdapterPlugin, working Credentials, ConnectionManager с open()/get_response()/cancel(), basic execute(), list_schemas macro.

3. Inherit from SQLAdapter — большинство getting работает out-of-box (cursor execute, ANSI SQL macros).

4. Override адаптер-specific только when default не работает (warehouse-specific SQL, non-standard types).

5. debug — gate в CI. Pre-deploy validation. exit code 0 = OK, 2 = fail.

6. Common failures: adapter not found (pip install), connection refused (warehouse not running), authentication failed (wrong creds), macros (override default).