From d53f1b97e47e36f610c90c3b5dde8594f190869e Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Thu, 12 Jun 2025 09:32:02 +0700 Subject: [PATCH 01/15] add is_auto_partition config --- .idea/dbt-doris.iml | 2 +- .idea/misc.xml | 5 ++++- .python-version | 1 + README.md | 1 + dbt/adapters/doris/connections.py | 16 ++++++++-------- dbt/adapters/doris/impl.py | 13 +++++++------ dbt/adapters/doris/relation.py | 2 +- dbt/include/doris/dbt_project.yml | 2 +- dbt/include/doris/macros/adapters/relation.sql | 3 ++- dev-requirements.txt | 6 ++---- main.py | 6 ++++++ setup.py | 6 +++--- 12 files changed, 37 insertions(+), 26 deletions(-) create mode 100644 .python-version create mode 100644 main.py diff --git a/.idea/dbt-doris.iml b/.idea/dbt-doris.iml index a3166ab..3310613 100644 --- a/.idea/dbt-doris.iml +++ b/.idea/dbt-doris.iml @@ -4,7 +4,7 @@ - + diff --git a/.idea/misc.xml b/.idea/misc.xml index d235ddd..85f482d 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -1,4 +1,7 @@ - + + + \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..e4fba21 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.12 diff --git a/README.md b/README.md index e701f32..7c7daf7 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ This adapter plugin follows [semantic versioning](https://semver.org/). The firs ## Getting Started #### Setting up Locally +- run `pip install --only-binary :all: typed-ast==1.5.5` - run `pip install -r dev-requirements.txt`. - cd directory into the `dbt-core` you'd like to be testing against and run `make dev`. diff --git a/dbt/adapters/doris/connections.py b/dbt/adapters/doris/connections.py index f19878b..6585ea5 100644 --- a/dbt/adapters/doris/connections.py +++ b/dbt/adapters/doris/connections.py @@ -24,11 +24,11 @@ import mysql.connector -from dbt import exceptions -from dbt.adapters.base import Credentials +from dbt.adapters import exceptions +from dbt_common.exceptions import DbtRuntimeError, DbtDatabaseError from dbt.adapters.sql import SQLConnectionManager -from dbt.contracts.connection import AdapterResponse, Connection, ConnectionState -from dbt.events import AdapterLogger +from dbt.adapters.contracts.connection import AdapterResponse, Connection, Credentials +from dbt.adapters.events.logging import AdapterLogger logger = AdapterLogger("doris") @@ -56,7 +56,7 @@ def unique_field(self) -> str: def __post_init__(self): if self.database is not None and self.database != self.schema: - raise exceptions.DbtRuntimeError( + raise DbtRuntimeError( f" schema: {self.schema} \n" f" database: {self.database} \n" f"On Doris, database must be omitted or have the same value as" @@ -132,12 +132,12 @@ def exception_handler(self, sql: str) -> ContextManager: yield except mysql.connector.DatabaseError as e: logger.debug(f"Doris database error: {e}, sql: {sql}") - raise exceptions.DbtDatabaseError(str(e)) from e + raise DbtDatabaseError(str(e)) from e except Exception as e: logger.debug(f"Error running SQL: {sql}") - if isinstance(e, exceptions.DbtRuntimeError): + if isinstance(e, DbtRuntimeError): raise e - raise exceptions.DbtRuntimeError(str(e)) from e + raise DbtRuntimeError(str(e)) from e @classmethod def begin(self): diff --git a/dbt/adapters/doris/impl.py b/dbt/adapters/doris/impl.py index ba6406d..980a631 100644 --- a/dbt/adapters/doris/impl.py +++ b/dbt/adapters/doris/impl.py @@ -38,7 +38,7 @@ ) import agate -import dbt.exceptions +import dbt_common.exceptions from dbt.adapters.base.impl import _expect_row_value, catch_as_completed from dbt.adapters.base.relation import InformationSchema, BaseRelation from dbt.adapters.doris.column import DorisColumn @@ -46,10 +46,10 @@ from dbt.adapters.doris.relation import DorisRelation from dbt.adapters.protocol import AdapterConfig from dbt.adapters.sql.impl import LIST_RELATIONS_MACRO_NAME, LIST_SCHEMAS_MACRO_NAME -from dbt.clients.agate_helper import table_from_rows +from dbt_common.clients.agate_helper import table_from_rows from dbt.contracts.graph.manifest import Manifest -from dbt.contracts.relation import RelationType -from dbt.utils import executor +from dbt.adapters.contracts.relation import RelationType +from dbt_common.utils import executor from dbt.adapters.doris.doris_column_item import DorisColumnItem @@ -74,6 +74,7 @@ class DorisConfig(AdapterConfig): partition_by_init: List[str] distributed_by: Tuple[str] buckets: int + is_auto_partition: bool properties: Dict[str, str] @@ -123,7 +124,7 @@ def list_relations_without_caching(self, schema_relation: DorisRelation) -> List relations = [] for row in results: if len(row) != 4: - raise dbt.exceptions.DbtRuntimeError( + raise dbt_common.exceptions.DbtRuntimeError( f"Invalid value from 'show table extended ...', " f"got {len(row)} values, expected 4" ) @@ -188,7 +189,7 @@ def _get_one_catalog( manifest: Manifest, ) -> agate.Table: if len(schemas) != 1: - dbt.exceptions.raise_compiler_error( + raise_compiler_error( f"Expected only one schema in Doris _get_one_catalog, found " f"{schemas}" ) diff --git a/dbt/adapters/doris/relation.py b/dbt/adapters/doris/relation.py index 17956d1..201ec80 100644 --- a/dbt/adapters/doris/relation.py +++ b/dbt/adapters/doris/relation.py @@ -21,7 +21,7 @@ from dataclasses import dataclass, field from dbt.adapters.base.relation import BaseRelation, Policy -from dbt.exceptions import DbtRuntimeError +from dbt_common.exceptions import DbtRuntimeError @dataclass diff --git a/dbt/include/doris/dbt_project.yml b/dbt/include/doris/dbt_project.yml index 337002a..4b52865 100644 --- a/dbt/include/doris/dbt_project.yml +++ b/dbt/include/doris/dbt_project.yml @@ -19,7 +19,7 @@ # under the License. name: dbt_doris -version: 0.3.4 +version: 0.3.5 config-version: 2 macro-paths: ["macros"] diff --git a/dbt/include/doris/macros/adapters/relation.sql b/dbt/include/doris/macros/adapters/relation.sql index 3b2950d..3becaae 100644 --- a/dbt/include/doris/macros/adapters/relation.sql +++ b/dbt/include/doris/macros/adapters/relation.sql @@ -24,11 +24,12 @@ {% macro doris__partition_by() -%} {% set cols = config.get('partition_by', validator=validation.any[list, basestring]) %} {% set partition_type = config.get('partition_type', 'RANGE') %} + {% set is_auto_partition = config.get('is_auto_partition', false) %} {% if cols is not none %} {%- if cols is string -%} {%- set cols = [cols] -%} {%- endif -%} - PARTITION BY {{ partition_type }} ( + {% if is_auto_partition is true %} AUTO {%- endif -%} PARTITION BY {{ partition_type }} ( {% for col in cols %} {{ col }}{% if not loop.last %},{% endif %} {% endfor %} diff --git a/dev-requirements.txt b/dev-requirements.txt index 5b48d70..aa4add8 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -1,7 +1,5 @@ -# install latest changes in dbt-core -git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-core&subdirectory=core -git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-tests-adapter&subdirectory=tests/adapter - +dbt-core==1.9.8 +dbt-tests-adapter black==22.3.0 bumpversion flake8 diff --git a/main.py b/main.py new file mode 100644 index 0000000..752d119 --- /dev/null +++ b/main.py @@ -0,0 +1,6 @@ +def main(): + print("Hello from dbt-doris!") + + +if __name__ == "__main__": + main() diff --git a/setup.py b/setup.py index b76ee20..b16f13c 100644 --- a/setup.py +++ b/setup.py @@ -3,8 +3,8 @@ package_name = "dbt-doris" # make sure this always matches dbt/adapters/{adapter}/__version__.py -package_version = "0.3.4" -dbt_core_version = "1.5.0" +package_version = "0.3.5" +dbt_core_version = "1.9.8" description = """The doris adapter plugin for dbt """ setup( @@ -18,7 +18,7 @@ packages=find_namespace_packages(include=["dbt", "dbt.*"]), include_package_data=True, install_requires=[ - "dbt-core~={}".format(dbt_core_version), + f"dbt-core>={dbt_core_version}", "mysql-connector-python>=8.0.0,<8.1", "urllib3~=1.0", ], From fe0b311d826b92d5d1c65ec3d444c2cee5744625 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Thu, 12 Jun 2025 10:34:07 +0700 Subject: [PATCH 02/15] add a new line in the AUTO keyword --- dbt/include/doris/macros/adapters/relation.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dbt/include/doris/macros/adapters/relation.sql b/dbt/include/doris/macros/adapters/relation.sql index 3becaae..1fbaaad 100644 --- a/dbt/include/doris/macros/adapters/relation.sql +++ b/dbt/include/doris/macros/adapters/relation.sql @@ -29,7 +29,7 @@ {%- if cols is string -%} {%- set cols = [cols] -%} {%- endif -%} - {% if is_auto_partition is true %} AUTO {%- endif -%} PARTITION BY {{ partition_type }} ( + {%- if is_auto_partition is true -%} AUTO {%- endif -%} PARTITION BY {{ partition_type }} ( {% for col in cols %} {{ col }}{% if not loop.last %},{% endif %} {% endfor %} From ea9a546126658e819dcf21e35a1c8228cb27bcf6 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Thu, 12 Jun 2025 10:35:08 +0700 Subject: [PATCH 03/15] add a new line in the AUTO keyword --- dbt/include/doris/macros/adapters/relation.sql | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dbt/include/doris/macros/adapters/relation.sql b/dbt/include/doris/macros/adapters/relation.sql index 1fbaaad..562c656 100644 --- a/dbt/include/doris/macros/adapters/relation.sql +++ b/dbt/include/doris/macros/adapters/relation.sql @@ -29,7 +29,8 @@ {%- if cols is string -%} {%- set cols = [cols] -%} {%- endif -%} - {%- if is_auto_partition is true -%} AUTO {%- endif -%} PARTITION BY {{ partition_type }} ( + {%- if is_auto_partition is true -%} AUTO {%- endif -%} + PARTITION BY {{ partition_type }} ( {% for col in cols %} {{ col }}{% if not loop.last %},{% endif %} {% endfor %} From 2c95153e728041c678a9a6f3289f051de9567ab2 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Thu, 12 Jun 2025 10:37:50 +0700 Subject: [PATCH 04/15] add a new line in the AUTO keyword --- dbt/include/doris/macros/adapters/relation.sql | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dbt/include/doris/macros/adapters/relation.sql b/dbt/include/doris/macros/adapters/relation.sql index 562c656..5693fea 100644 --- a/dbt/include/doris/macros/adapters/relation.sql +++ b/dbt/include/doris/macros/adapters/relation.sql @@ -29,8 +29,8 @@ {%- if cols is string -%} {%- set cols = [cols] -%} {%- endif -%} - {%- if is_auto_partition is true -%} AUTO {%- endif -%} - PARTITION BY {{ partition_type }} ( + {% if is_auto_partition is true %} AUTO PARTITION BY {% else %} PARTITION BY {% endif %} + {{ partition_type }} ( {% for col in cols %} {{ col }}{% if not loop.last %},{% endif %} {% endfor %} From 90c32a2b0a44f49d17f2b80c7211b9cc04bf6c6c Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Thu, 12 Jun 2025 10:44:11 +0700 Subject: [PATCH 05/15] bump adapter version to 0.3.5 --- dbt/adapters/doris/__version__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dbt/adapters/doris/__version__.py b/dbt/adapters/doris/__version__.py index ba5a23c..52ba423 100644 --- a/dbt/adapters/doris/__version__.py +++ b/dbt/adapters/doris/__version__.py @@ -22,4 +22,4 @@ # this 'version' must be set !!! # otherwise the adapters will not be found after the 'dbt init xxx' command -version = "0.3.4" +version = "0.3.5" From 876423588fd7a2962c10a742e6d14b8efc7fb734 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Fri, 13 Jun 2025 15:44:52 +0700 Subject: [PATCH 06/15] on_schema_change --- .../doris/macros/materializations/incremental/incremental.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 60b0a34..58f4647 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -27,7 +27,7 @@ {% set to_drop = [] %} {#-- append or no unique key --#} - + {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %} {% if not unique_key or strategy == 'append' %} {#-- create table first --#} From 83eae47c52a38ce53921c29f98b28573bd3de121 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Fri, 13 Jun 2025 16:20:03 +0700 Subject: [PATCH 07/15] on_schema_change 1 --- .../incremental/on_schema_change.sql | 144 ++++++++++++++++++ 1 file changed, 144 insertions(+) create mode 100644 dbt/include/doris/macros/materializations/incremental/on_schema_change.sql diff --git a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql new file mode 100644 index 0000000..76fe372 --- /dev/null +++ b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql @@ -0,0 +1,144 @@ +{% macro incremental_validate_on_schema_change(on_schema_change, default='ignore') %} + + {% if on_schema_change not in ['sync_all_columns', 'append_new_columns', 'fail', 'ignore'] %} + + {% set log_message = 'Invalid value for on_schema_change (%s) specified. Setting default value of %s.' % (on_schema_change, default) %} + {% do log(log_message) %} + + {{ return(default) }} + + {% else %} + + {{ return(on_schema_change) }} + + {% endif %} + +{% endmacro %} + + +{% macro check_for_schema_changes(source_relation, target_relation) %} + + {% set schema_changed = False %} + + {%- set source_columns = adapter.get_columns_in_relation(source_relation) -%} + {%- set target_columns = adapter.get_columns_in_relation(target_relation) -%} + {%- set source_not_in_target = diff_columns(source_columns, target_columns) -%} + {%- set target_not_in_source = diff_columns(target_columns, source_columns) -%} + + {% set new_target_types = diff_column_data_types(source_columns, target_columns) %} + + {% if source_not_in_target != [] %} + {% set schema_changed = True %} + {% elif target_not_in_source != [] or new_target_types != [] %} + {% set schema_changed = True %} + {% elif new_target_types != [] %} + {% set schema_changed = True %} + {% endif %} + + {% set changes_dict = { + 'schema_changed': schema_changed, + 'source_not_in_target': source_not_in_target, + 'target_not_in_source': target_not_in_source, + 'source_columns': source_columns, + 'target_columns': target_columns, + 'new_target_types': new_target_types + } %} + + {% set msg %} + In {{ target_relation }}: + Schema changed: {{ schema_changed }} + Source columns not in target: {{ source_not_in_target }} + Target columns not in source: {{ target_not_in_source }} + New column types: {{ new_target_types }} + {% endset %} + + {% do log(msg) %} + + {{ return(changes_dict) }} + +{% endmacro %} + + +{% macro sync_column_schemas(on_schema_change, target_relation, schema_changes_dict) %} + + {%- set add_to_target_arr = schema_changes_dict['source_not_in_target'] -%} + + {%- if on_schema_change == 'append_new_columns'-%} + {%- if add_to_target_arr | length > 0 -%} + {%- do alter_relation_add_remove_columns(target_relation, add_to_target_arr, none) -%} + {%- endif -%} + + {% elif on_schema_change == 'sync_all_columns' %} + {%- set remove_from_target_arr = schema_changes_dict['target_not_in_source'] -%} + {%- set new_target_types = schema_changes_dict['new_target_types'] -%} + + {% if add_to_target_arr | length > 0 or remove_from_target_arr | length > 0 %} + {%- do alter_relation_add_remove_columns(target_relation, add_to_target_arr, remove_from_target_arr) -%} + {% endif %} + + {% if new_target_types != [] %} + {% for ntt in new_target_types %} + {% set column_name = ntt['column_name'] %} + {% set new_type = ntt['new_type'] %} + {% do alter_column_type(target_relation, column_name, new_type) %} + {% endfor %} + {% endif %} + + {% endif %} + + {% set schema_change_message %} + In {{ target_relation }}: + Schema change approach: {{ on_schema_change }} + Columns added: {{ add_to_target_arr }} + Columns removed: {{ remove_from_target_arr }} + Data types changed: {{ new_target_types }} + {% endset %} + + {% do log(schema_change_message) %} + +{% endmacro %} + + +{% macro process_schema_changes(on_schema_change, source_relation, target_relation) %} + + {% if on_schema_change == 'ignore' %} + + {{ return({}) }} + + {% else %} + + {% set schema_changes_dict = check_for_schema_changes(source_relation, target_relation) %} + + {% if schema_changes_dict['schema_changed'] %} + + {% if on_schema_change == 'fail' %} + + {% set fail_msg %} + The source and target schemas on this incremental model are out of sync! + They can be reconciled in several ways: + - set the `on_schema_change` config to either append_new_columns or sync_all_columns, depending on your situation. + - Re-run the incremental model with `full_refresh: True` to update the target schema. + - update the schema manually and re-run the process. + + Additional troubleshooting context: + Source columns not in target: {{ schema_changes_dict['source_not_in_target'] }} + Target columns not in source: {{ schema_changes_dict['target_not_in_source'] }} + New column types: {{ schema_changes_dict['new_target_types'] }} + {% endset %} + + {% do exceptions.raise_compiler_error(fail_msg) %} + + {# -- unless we ignore, run the sync operation per the config #} + {% else %} + + {% do sync_column_schemas(on_schema_change, target_relation, schema_changes_dict) %} + + {% endif %} + + {% endif %} + + {{ return(schema_changes_dict['source_columns']) }} + + {% endif %} + +{% endmacro %} From ec01a4bfca7f4e5458ef98bbf3c5caab64ef9d68 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Fri, 13 Jun 2025 16:27:07 +0700 Subject: [PATCH 08/15] on_schema_change 2 --- .../doris/macros/materializations/incremental/incremental.sql | 2 +- .../macros/materializations/incremental/on_schema_change.sql | 2 -- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 58f4647..2e3e88c 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -27,7 +27,7 @@ {% set to_drop = [] %} {#-- append or no unique key --#} - {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %} + {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='fail') %} {% if not unique_key or strategy == 'append' %} {#-- create table first --#} diff --git a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql index 76fe372..e6b67ab 100644 --- a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql +++ b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql @@ -1,10 +1,8 @@ {% macro incremental_validate_on_schema_change(on_schema_change, default='ignore') %} {% if on_schema_change not in ['sync_all_columns', 'append_new_columns', 'fail', 'ignore'] %} - {% set log_message = 'Invalid value for on_schema_change (%s) specified. Setting default value of %s.' % (on_schema_change, default) %} {% do log(log_message) %} - {{ return(default) }} {% else %} From 889f19b6a25a8a2c3267454e253a033d6ee06edf Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Fri, 13 Jun 2025 16:48:11 +0700 Subject: [PATCH 09/15] on_schema_change 3 --- .../macros/materializations/incremental/incremental.sql | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 2e3e88c..848ed80 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -79,6 +79,11 @@ {% do adapter.expand_target_column_types( from_relation=tmp_relation, to_relation=target_relation) %} + {#-- Process schema changes. Returns dict of changes if successful. Use source columns for upserting/merging --#} + {% set dest_columns = process_schema_changes(on_schema_change, temp_relation, existing_relation) %} + {% if not dest_columns %} + {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %} + {% endif %} {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=unique_key) %} {% endif %} {% else %} From fc24b7f77e5a95e008904caa545f309faf62c257 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Sat, 14 Jun 2025 09:39:36 +0700 Subject: [PATCH 10/15] on_schema_change 4 --- .../incremental/incremental.sql | 5 - .../incremental/on_schema_change.sql | 142 ------------------ 2 files changed, 147 deletions(-) delete mode 100644 dbt/include/doris/macros/materializations/incremental/on_schema_change.sql diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 848ed80..2e3e88c 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -79,11 +79,6 @@ {% do adapter.expand_target_column_types( from_relation=tmp_relation, to_relation=target_relation) %} - {#-- Process schema changes. Returns dict of changes if successful. Use source columns for upserting/merging --#} - {% set dest_columns = process_schema_changes(on_schema_change, temp_relation, existing_relation) %} - {% if not dest_columns %} - {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %} - {% endif %} {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=unique_key) %} {% endif %} {% else %} diff --git a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql deleted file mode 100644 index e6b67ab..0000000 --- a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql +++ /dev/null @@ -1,142 +0,0 @@ -{% macro incremental_validate_on_schema_change(on_schema_change, default='ignore') %} - - {% if on_schema_change not in ['sync_all_columns', 'append_new_columns', 'fail', 'ignore'] %} - {% set log_message = 'Invalid value for on_schema_change (%s) specified. Setting default value of %s.' % (on_schema_change, default) %} - {% do log(log_message) %} - {{ return(default) }} - - {% else %} - - {{ return(on_schema_change) }} - - {% endif %} - -{% endmacro %} - - -{% macro check_for_schema_changes(source_relation, target_relation) %} - - {% set schema_changed = False %} - - {%- set source_columns = adapter.get_columns_in_relation(source_relation) -%} - {%- set target_columns = adapter.get_columns_in_relation(target_relation) -%} - {%- set source_not_in_target = diff_columns(source_columns, target_columns) -%} - {%- set target_not_in_source = diff_columns(target_columns, source_columns) -%} - - {% set new_target_types = diff_column_data_types(source_columns, target_columns) %} - - {% if source_not_in_target != [] %} - {% set schema_changed = True %} - {% elif target_not_in_source != [] or new_target_types != [] %} - {% set schema_changed = True %} - {% elif new_target_types != [] %} - {% set schema_changed = True %} - {% endif %} - - {% set changes_dict = { - 'schema_changed': schema_changed, - 'source_not_in_target': source_not_in_target, - 'target_not_in_source': target_not_in_source, - 'source_columns': source_columns, - 'target_columns': target_columns, - 'new_target_types': new_target_types - } %} - - {% set msg %} - In {{ target_relation }}: - Schema changed: {{ schema_changed }} - Source columns not in target: {{ source_not_in_target }} - Target columns not in source: {{ target_not_in_source }} - New column types: {{ new_target_types }} - {% endset %} - - {% do log(msg) %} - - {{ return(changes_dict) }} - -{% endmacro %} - - -{% macro sync_column_schemas(on_schema_change, target_relation, schema_changes_dict) %} - - {%- set add_to_target_arr = schema_changes_dict['source_not_in_target'] -%} - - {%- if on_schema_change == 'append_new_columns'-%} - {%- if add_to_target_arr | length > 0 -%} - {%- do alter_relation_add_remove_columns(target_relation, add_to_target_arr, none) -%} - {%- endif -%} - - {% elif on_schema_change == 'sync_all_columns' %} - {%- set remove_from_target_arr = schema_changes_dict['target_not_in_source'] -%} - {%- set new_target_types = schema_changes_dict['new_target_types'] -%} - - {% if add_to_target_arr | length > 0 or remove_from_target_arr | length > 0 %} - {%- do alter_relation_add_remove_columns(target_relation, add_to_target_arr, remove_from_target_arr) -%} - {% endif %} - - {% if new_target_types != [] %} - {% for ntt in new_target_types %} - {% set column_name = ntt['column_name'] %} - {% set new_type = ntt['new_type'] %} - {% do alter_column_type(target_relation, column_name, new_type) %} - {% endfor %} - {% endif %} - - {% endif %} - - {% set schema_change_message %} - In {{ target_relation }}: - Schema change approach: {{ on_schema_change }} - Columns added: {{ add_to_target_arr }} - Columns removed: {{ remove_from_target_arr }} - Data types changed: {{ new_target_types }} - {% endset %} - - {% do log(schema_change_message) %} - -{% endmacro %} - - -{% macro process_schema_changes(on_schema_change, source_relation, target_relation) %} - - {% if on_schema_change == 'ignore' %} - - {{ return({}) }} - - {% else %} - - {% set schema_changes_dict = check_for_schema_changes(source_relation, target_relation) %} - - {% if schema_changes_dict['schema_changed'] %} - - {% if on_schema_change == 'fail' %} - - {% set fail_msg %} - The source and target schemas on this incremental model are out of sync! - They can be reconciled in several ways: - - set the `on_schema_change` config to either append_new_columns or sync_all_columns, depending on your situation. - - Re-run the incremental model with `full_refresh: True` to update the target schema. - - update the schema manually and re-run the process. - - Additional troubleshooting context: - Source columns not in target: {{ schema_changes_dict['source_not_in_target'] }} - Target columns not in source: {{ schema_changes_dict['target_not_in_source'] }} - New column types: {{ schema_changes_dict['new_target_types'] }} - {% endset %} - - {% do exceptions.raise_compiler_error(fail_msg) %} - - {# -- unless we ignore, run the sync operation per the config #} - {% else %} - - {% do sync_column_schemas(on_schema_change, target_relation, schema_changes_dict) %} - - {% endif %} - - {% endif %} - - {{ return(schema_changes_dict['source_columns']) }} - - {% endif %} - -{% endmacro %} From 84e74bd11c978bcb9f14d94db80c6a50a504247a Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Sat, 14 Jun 2025 10:08:40 +0700 Subject: [PATCH 11/15] on_schema_change 5 --- dbt/adapters/doris/impl.py | 35 +++++++++++++ dbt/include/doris/macros/adapters/columns.sql | 51 +++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/dbt/adapters/doris/impl.py b/dbt/adapters/doris/impl.py index 980a631..6702413 100644 --- a/dbt/adapters/doris/impl.py +++ b/dbt/adapters/doris/impl.py @@ -203,6 +203,41 @@ def timestamp_add_sql(self, add_to: str, number: int = 1, interval: str = "hour" # and might even be the SQL standard's intention. return f"{add_to} + interval {number} {interval}" + def alter_column_type(self, relation, column_name, new_column_type): + """Alter column type in Doris.""" + relation_name = self._make_relation_name_for_query(relation) + self.execute_macro( + "doris__alter_column_type", + kwargs={"relation": relation_name, "column_name": column_name, "new_column_type": new_column_type} + ) + + def add_column(self, relation, column_name, column_type): + """Add a column to Doris table.""" + relation_name = self._make_relation_name_for_query(relation) + self.execute_macro( + "doris__add_column", + kwargs={"relation": relation_name, "column_name": column_name, "column_type": column_type} + ) + + def drop_column(self, relation, column_name): + """Drop column from Doris table.""" + relation_name = self._make_relation_name_for_query(relation) + self.execute_macro( + "doris__drop_column", + kwargs={"relation": relation_name, "column_name": column_name} + ) + + def rename_column(self, relation, old_column_name, new_column_name): + """Rename column in Doris table.""" + relation_name = self._make_relation_name_for_query(relation) + self.execute_macro( + "doris__rename_column", + kwargs={ + "relation": relation_name, + "old_column_name": old_column_name, + "new_column_name": new_column_name + } + ) @classmethod def render_raw_columns_constraints(cls, raw_columns: Dict[str, Dict[str, Any]]) -> List: diff --git a/dbt/include/doris/macros/adapters/columns.sql b/dbt/include/doris/macros/adapters/columns.sql index b1a834c..47b6bdf 100644 --- a/dbt/include/doris/macros/adapters/columns.sql +++ b/dbt/include/doris/macros/adapters/columns.sql @@ -38,3 +38,54 @@ where table_schema = '{{ relation.schema }}' {{ return(columns_and_constraints("view")) }} {%- endmacro %} +{% macro doris_alter_column_type(relation, column_name, new_column_type) %} +{% set sql %} +ALTER TABLE {{ relation }} MODIFY COLUMN {{ column_name }} {{ new_column_type }}; +{% endset %} +{% do run_query(sql) %} +{% do log("Column '" ~ column_name ~ "' type changed to '" ~ new_column_type ~ "' in " ~ relation, info=true) %} +{% endmacro %} + +{% macro doris_add_column(relation, column_name, column_type, column_position='', column_default='', column_comment='') %} +{% set sql %} +ALTER TABLE {{ relation }} ADD COLUMN {{ column_name }} {{ column_type }} + {% if column_default != '' %}DEFAULT {{ column_default }}{% endif %} + {% if column_comment != '' %}COMMENT '{{ column_comment }}'{% endif %} + {% if column_position != '' %}{{ column_position }}{% endif %}; +{% endset %} +{% do run_query(sql) %} +{% do log("Column '" ~ column_name ~ "' added to " ~ relation, info=true) %} +{% endmacro %} + +{% macro doris_drop_column(relation, column_name) %} +{% set sql %} +ALTER TABLE {{ relation }} DROP COLUMN {{ column_name }}; +{% endset %} +{% do run_query(sql) %} +{% do log("Column '" ~ column_name ~ "' dropped from " ~ relation, info=true) %} +{% endmacro %} + +{% macro doris_rename_column(relation, old_column_name, new_column_name) %} +{% set sql %} +ALTER TABLE {{ relation }} RENAME COLUMN {{ old_column_name }} {{ new_column_name }}; +{% endset %} +{% do run_query(sql) %} +{% do log("Column '" ~ old_column_name ~ "' renamed to '" ~ new_column_name ~ "' in " ~ relation, info=true) %} +{% endmacro %} + + +{% macro doris__alter_column_type(relation, column_name, new_column_type) %} +{{ doris_alter_column_type(relation, column_name, new_column_type) }} +{% endmacro %} + +{% macro doris__add_column(relation, column_name, column_type) %} +{{ doris_add_column(relation, column_name, column_type) }} +{% endmacro %} + +{% macro doris__drop_column(relation, column_name) %} +{{ doris_drop_column(relation, column_name) }} +{% endmacro %} + +{% macro doris__rename_column(relation, old_column_name, new_column_name) %} +{{ doris_rename_column(relation, old_column_name, new_column_name) }} +{% endmacro %} \ No newline at end of file From 2efb25681eaae8664bc8d22eeee7e246f46a607d Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Sat, 14 Jun 2025 10:34:25 +0700 Subject: [PATCH 12/15] on_schema_change 6 --- dbt/adapters/doris/column.py | 4 +++ dbt/adapters/doris/impl.py | 12 +++---- dbt/include/doris/macros/adapters/columns.sql | 4 --- .../incremental/incremental.sql | 34 ++++++++++++++++++- 4 files changed, 41 insertions(+), 13 deletions(-) diff --git a/dbt/adapters/doris/column.py b/dbt/adapters/doris/column.py index e5f8da0..82caca2 100644 --- a/dbt/adapters/doris/column.py +++ b/dbt/adapters/doris/column.py @@ -29,5 +29,9 @@ class DorisColumn(Column): def quoted(self) -> str: return "`{}`".format(self.column) + @classmethod + def string_type(cls, size: int) -> str: + return "VARCHAR({})".format(size) + def __repr__(self) -> str: return f"" diff --git a/dbt/adapters/doris/impl.py b/dbt/adapters/doris/impl.py index 6702413..519985b 100644 --- a/dbt/adapters/doris/impl.py +++ b/dbt/adapters/doris/impl.py @@ -205,35 +205,31 @@ def timestamp_add_sql(self, add_to: str, number: int = 1, interval: str = "hour" def alter_column_type(self, relation, column_name, new_column_type): """Alter column type in Doris.""" - relation_name = self._make_relation_name_for_query(relation) self.execute_macro( "doris__alter_column_type", - kwargs={"relation": relation_name, "column_name": column_name, "new_column_type": new_column_type} + kwargs={"relation": relation, "column_name": column_name, "new_column_type": new_column_type} ) def add_column(self, relation, column_name, column_type): """Add a column to Doris table.""" - relation_name = self._make_relation_name_for_query(relation) self.execute_macro( "doris__add_column", - kwargs={"relation": relation_name, "column_name": column_name, "column_type": column_type} + kwargs={"relation": relation, "column_name": column_name, "column_type": column_type} ) def drop_column(self, relation, column_name): """Drop column from Doris table.""" - relation_name = self._make_relation_name_for_query(relation) self.execute_macro( "doris__drop_column", - kwargs={"relation": relation_name, "column_name": column_name} + kwargs={"relation": relation, "column_name": column_name} ) def rename_column(self, relation, old_column_name, new_column_name): """Rename column in Doris table.""" - relation_name = self._make_relation_name_for_query(relation) self.execute_macro( "doris__rename_column", kwargs={ - "relation": relation_name, + "relation": relation, "old_column_name": old_column_name, "new_column_name": new_column_name } diff --git a/dbt/include/doris/macros/adapters/columns.sql b/dbt/include/doris/macros/adapters/columns.sql index 47b6bdf..92c677f 100644 --- a/dbt/include/doris/macros/adapters/columns.sql +++ b/dbt/include/doris/macros/adapters/columns.sql @@ -13,10 +13,6 @@ where table_schema = '{{ relation.schema }}' {{ return(sql_convert_columns_in_relation(table)) }} {%- endmacro %} -{% macro doris__alter_column_type(relation,column_name,new_column_type) -%} -'''Changes column name or data type''' -{% endmacro %} - {% macro columns_and_constraints(table_type="table") %} {# loop through user_provided_columns to create DDL with data types and constraints #} {%- set raw_column_constraints = adapter.render_raw_columns_constraints(raw_columns=model['columns']) -%} diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 2e3e88c..28fc4ff 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -27,7 +27,39 @@ {% set to_drop = [] %} {#-- append or no unique key --#} - {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='fail') %} + {# -- Check if we need to make schema changes (only applies if the table already exists) #} + {% if existing_relation is not none and not existing_relation.is_view and not full_refresh_mode %} + {% set tmp_table_sql = sql %} + {% do run_query(create_table_as(True, tmp_relation, sql)) %} + {% do adapter.expand_target_column_types(from_relation=tmp_relation, to_relation=existing_relation) %} + {% set target_columns = adapter.get_columns_in_relation(existing_relation) %} + {% set tmp_columns = adapter.get_columns_in_relation(tmp_relation) %} + + {# -- Compare the schemas #} + {% set target_cols_csv = target_columns | map(attribute='name') | join(', ') %} + {% set tmp_cols_csv = tmp_columns | map(attribute='name') | join(', ') %} + {% set missing_cols = [] %} + {% for col in tmp_columns %} + {% if col.name | lower not in target_columns | map(attribute='name') | map('lower') | list %} + {% do missing_cols.append(col) %} + {% endif %} + {% endfor %} + + {# -- Apply schema changes based on the on_schema_change setting #} + {% if missing_cols and on_schema_change != 'ignore' %} + {% if on_schema_change == 'fail' %} + {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ missing_cols | map(attribute='name') | join(', ') ~ '. Add columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} + {% elif on_schema_change == 'append_new_columns' %} + {% do log("Adding new columns: " ~ missing_cols | map(attribute='name') | join(', '), info=True) %} + {% for col in missing_cols %} + {% set sql_add_col %} + {{ doris_add_column(existing_relation, col.name, col.data_type) }} + {% endset %} + {% do run_query(sql_add_col) %} + {% endfor %} + {% endif %} + {% endif %} + {% endif %} {% if not unique_key or strategy == 'append' %} {#-- create table first --#} From d3059d0cff9f79e0c518b4546d1f9fa99246cbd7 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Sat, 14 Jun 2025 10:44:22 +0700 Subject: [PATCH 13/15] on_schema_change 7 refactor handle_schema_evolution --- .../incremental/incremental.sql | 36 ++----------------- .../incremental/on_schema_change.sql | 35 ++++++++++++++++++ 2 files changed, 37 insertions(+), 34 deletions(-) create mode 100644 dbt/include/doris/macros/materializations/incremental/on_schema_change.sql diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index 28fc4ff..dc37bcd 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -27,40 +27,6 @@ {% set to_drop = [] %} {#-- append or no unique key --#} - {# -- Check if we need to make schema changes (only applies if the table already exists) #} - {% if existing_relation is not none and not existing_relation.is_view and not full_refresh_mode %} - {% set tmp_table_sql = sql %} - {% do run_query(create_table_as(True, tmp_relation, sql)) %} - {% do adapter.expand_target_column_types(from_relation=tmp_relation, to_relation=existing_relation) %} - {% set target_columns = adapter.get_columns_in_relation(existing_relation) %} - {% set tmp_columns = adapter.get_columns_in_relation(tmp_relation) %} - - {# -- Compare the schemas #} - {% set target_cols_csv = target_columns | map(attribute='name') | join(', ') %} - {% set tmp_cols_csv = tmp_columns | map(attribute='name') | join(', ') %} - {% set missing_cols = [] %} - {% for col in tmp_columns %} - {% if col.name | lower not in target_columns | map(attribute='name') | map('lower') | list %} - {% do missing_cols.append(col) %} - {% endif %} - {% endfor %} - - {# -- Apply schema changes based on the on_schema_change setting #} - {% if missing_cols and on_schema_change != 'ignore' %} - {% if on_schema_change == 'fail' %} - {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ missing_cols | map(attribute='name') | join(', ') ~ '. Add columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} - {% elif on_schema_change == 'append_new_columns' %} - {% do log("Adding new columns: " ~ missing_cols | map(attribute='name') | join(', '), info=True) %} - {% for col in missing_cols %} - {% set sql_add_col %} - {{ doris_add_column(existing_relation, col.name, col.data_type) }} - {% endset %} - {% do run_query(sql_add_col) %} - {% endfor %} - {% endif %} - {% endif %} - {% endif %} - {% if not unique_key or strategy == 'append' %} {#-- create table first --#} {% if existing_relation is none %} @@ -80,6 +46,7 @@ {% else %} {% do to_drop.append(tmp_relation) %} {% do run_query(create_table_as(True, tmp_relation, sql)) %} + {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=none) %} {% endif %} {#-- insert overwrite --#} @@ -107,6 +74,7 @@ {% endif %} {#-- create temp duplicate table for this incremental task --#} {% do run_query(create_table_as(True, tmp_relation, sql)) %} + {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} {% do to_drop.append(tmp_relation) %} {% do adapter.expand_target_column_types( from_relation=tmp_relation, diff --git a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql new file mode 100644 index 0000000..58e341a --- /dev/null +++ b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql @@ -0,0 +1,35 @@ +{% macro handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} + +{% if existing_relation is not none and not existing_relation.is_view and not full_refresh_mode %} + {% set tmp_table_sql = sql %} + {% do adapter.expand_target_column_types(from_relation=tmp_relation, to_relation=existing_relation) %} + {% set target_columns = adapter.get_columns_in_relation(existing_relation) %} + {% set tmp_columns = adapter.get_columns_in_relation(tmp_relation) %} + + {# -- Compare the schemas #} + {% set target_cols_csv = target_columns | map(attribute='name') | join(', ') %} + {% set tmp_cols_csv = tmp_columns | map(attribute='name') | join(', ') %} + {% set missing_cols = [] %} + {% for col in tmp_columns %} + {% if col.name | lower not in target_columns | map(attribute='name') | map('lower') | list %} + {% do missing_cols.append(col) %} + {% endif %} + {% endfor %} + + {# -- Apply schema changes based on the on_schema_change setting #} + {% if missing_cols and on_schema_change != 'ignore' %} + {% if on_schema_change == 'fail' %} + {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ missing_cols | map(attribute='name') | join(', ') ~ '. Add columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} + {% elif on_schema_change == 'append_new_columns' %} + {% do log("Adding new columns: " ~ missing_cols | map(attribute='name') | join(', '), info=True) %} + {% for col in missing_cols %} + {% set sql_add_col %} + {{ doris_add_column(existing_relation, col.name, col.data_type) }} + {% endset %} + {% do run_query(sql_add_col) %} + {% endfor %} + {% endif %} + {% endif %} +{% endif %} + +{% endmacro %} \ No newline at end of file From d18969a7eeefa89a15fd91eaacaa52a1b18fb693 Mon Sep 17 00:00:00 2001 From: anhtd1 Date: Sat, 14 Jun 2025 11:57:33 +0700 Subject: [PATCH 14/15] on_schema_change 8 refactor handle_schema_evolution --- .../incremental/incremental.sql | 107 +++++++++--------- .../incremental/on_schema_change.sql | 45 ++++++-- pyproject.toml | 10 ++ setup.py | 2 +- uv.lock | 36 ++++++ 5 files changed, 138 insertions(+), 62 deletions(-) create mode 100644 pyproject.toml create mode 100644 uv.lock diff --git a/dbt/include/doris/macros/materializations/incremental/incremental.sql b/dbt/include/doris/macros/materializations/incremental/incremental.sql index dc37bcd..e0b1191 100644 --- a/dbt/include/doris/macros/materializations/incremental/incremental.sql +++ b/dbt/include/doris/macros/materializations/incremental/incremental.sql @@ -28,72 +28,71 @@ {#-- append or no unique key --#} {% if not unique_key or strategy == 'append' %} - {#-- create table first --#} - {% if existing_relation is none %} - {% set build_sql = doris__create_table_as(False, target_relation, sql) %} - {% elif existing_relation.is_view or full_refresh_mode %} - {#-- backup table is new table ,exchange table backup and old table #} - {% set backup_identifier = existing_relation.identifier ~ "__dbt_backup" %} - {% set backup_relation = existing_relation.incorporate(path={"identifier": backup_identifier}) %} - {% do adapter.drop_relation(backup_relation) %} {#-- likes 'drop table if exists ... ' --#} - {% set run_sql = doris__create_table_as(False, backup_relation, sql) %} - {% call statement("run_sql") %} - {{ run_sql }} - {% endcall %} - {% do exchange_relation(target_relation, backup_relation, True) %} - {% set build_sql = "select 'hello doris'" %} - {#-- append data --#} - {% else %} - {% do to_drop.append(tmp_relation) %} - {% do run_query(create_table_as(True, tmp_relation, sql)) %} - {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} - {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=none) %} - {% endif %} + {#-- create table first --#} + {% if existing_relation is none %} + {% set build_sql = doris__create_table_as(False, target_relation, sql) %} + {% elif existing_relation.is_view or full_refresh_mode %} + {#-- backup table is new table ,exchange table backup and old table #} + {% set backup_identifier = existing_relation.identifier ~ "__dbt_backup" %} + {% set backup_relation = existing_relation.incorporate(path={"identifier": backup_identifier}) %} + {% do log("Drop relation: " ~ backup_relation, info=True) %} + {% do adapter.drop_relation(backup_relation) %} {#-- likes 'drop table if exists ... ' --#} + {% set run_sql = doris__create_table_as(False, backup_relation, sql) %} + {% call statement("run_sql") %} + {{ run_sql }} + {% endcall %} + {% do exchange_relation(target_relation, backup_relation, True) %} + {% set build_sql = "select 'hello doris'" %} + {#-- append data --#} + {% else %} + {% do to_drop.append(tmp_relation) %} + {% do run_query(create_table_as(True, tmp_relation, sql)) %} + {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} + {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=none) %} + {% endif %} {#-- insert overwrite --#} {% elif strategy == 'insert_overwrite' %} - {#-- create table first --#} - {% if existing_relation is none %} - {% set build_sql = doris__create_unique_table_as(False, target_relation, sql) %} - {#-- insert data refresh --#} - {% elif existing_relation.is_view or full_refresh_mode %} - {#-- backup table is new table ,exchange table backup and old table #} - {% set backup_identifier = existing_relation.identifier ~ "__dbt_backup" %} - {% set backup_relation = existing_relation.incorporate(path={"identifier": backup_identifier}) %} - {% do adapter.drop_relation(backup_relation) %} {#-- likes 'drop table if exists ... ' --#} - {% set run_sql = doris__create_unique_table_as(False, backup_relation, sql) %} - {% call statement("run_sql") %} - {{ run_sql }} - {% endcall %} - {% do exchange_relation(target_relation, backup_relation, True) %} - {% set build_sql = "select 'hello doris'" %} - {#-- append data --#} - {% else %} - {#-- check doris unique table --#} - {% if not is_unique_model(target_relation) %} - {% do exceptions.raise_compiler_error("doris table:"~ target_relation ~ ", model must be 'UNIQUE'" ) %} - {% endif %} - {#-- create temp duplicate table for this incremental task --#} - {% do run_query(create_table_as(True, tmp_relation, sql)) %} - {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} - {% do to_drop.append(tmp_relation) %} - {% do adapter.expand_target_column_types( - from_relation=tmp_relation, - to_relation=target_relation) %} - {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=unique_key) %} - {% endif %} + {#-- create table first --#} + {% if existing_relation is none %} + {% set build_sql = doris__create_unique_table_as(False, target_relation, sql) %} + {#-- insert data refresh --#} + {% elif existing_relation.is_view or full_refresh_mode %} + {#-- backup table is new table ,exchange table backup and old table #} + {% set backup_identifier = existing_relation.identifier ~ "__dbt_backup" %} + {% set backup_relation = existing_relation.incorporate(path={"identifier": backup_identifier}) %} + {% do adapter.drop_relation(backup_relation) %} {#-- likes 'drop table if exists ... ' --#} + {% set run_sql = doris__create_unique_table_as(False, backup_relation, sql) %} + {% call statement("run_sql") %} + {{ run_sql }} + {% endcall %} + {% do exchange_relation(target_relation, backup_relation, True) %} + {% set build_sql = "select 'hello doris'" %} + {#-- append data --#} + {% else %} + {#-- check doris unique table --#} + {% if not is_unique_model(target_relation) %} + {% do exceptions.raise_compiler_error("doris table:"~ target_relation ~ ", model must be 'UNIQUE'" ) %} + {% endif %} + {#-- create temp duplicate table for this incremental task --#} + {% do run_query(create_table_as(True, tmp_relation, sql)) %} + {% do handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} + {% do to_drop.append(tmp_relation) %} + {% do adapter.expand_target_column_types(from_relation=tmp_relation, to_relation=target_relation) %} + {% set build_sql = tmp_insert(tmp_relation, target_relation, unique_key=unique_key) %} + {% endif %} {% else %} - {#-- never --#} + {#-- never --#} {% endif %} {% call statement("main") %} - {{ build_sql }} + {{ build_sql }} {% endcall %} {#-- {% do persist_docs(target_relation, model) %} #} {{ run_hooks(post_hooks, inside_transaction=True) }} {% do adapter.commit() %} {% for rel in to_drop %} - {% do doris__drop_relation(rel) %} + {% do doris__drop_relation(rel) %} {% endfor %} {{ run_hooks(post_hooks, inside_transaction=False) }} {{ return({'relations': [target_relation]}) }} diff --git a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql index 58e341a..3dcee3e 100644 --- a/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql +++ b/dbt/include/doris/macros/materializations/incremental/on_schema_change.sql @@ -1,35 +1,66 @@ -{% macro handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation) %} +{% macro handle_schema_evolution(existing_relation, full_refresh_mode, tmp_relation, on_schema_change) %} {% if existing_relation is not none and not existing_relation.is_view and not full_refresh_mode %} {% set tmp_table_sql = sql %} + {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %} {% do adapter.expand_target_column_types(from_relation=tmp_relation, to_relation=existing_relation) %} {% set target_columns = adapter.get_columns_in_relation(existing_relation) %} {% set tmp_columns = adapter.get_columns_in_relation(tmp_relation) %} {# -- Compare the schemas #} - {% set target_cols_csv = target_columns | map(attribute='name') | join(', ') %} - {% set tmp_cols_csv = tmp_columns | map(attribute='name') | join(', ') %} + {% set target_cols_csv = target_columns | map(attribute='column') | join(', ') %} + {% set tmp_cols_csv = tmp_columns | map(attribute='column') | join(', ') %} + {% set missing_cols = [] %} + {% set changed_cols = [] %} + + {# -- Identify missing columns in the target relation #} {% for col in tmp_columns %} - {% if col.name | lower not in target_columns | map(attribute='name') | map('lower') | list %} + {% if col.column | lower not in target_columns | map(attribute='column') | map('lower') | list %} {% do missing_cols.append(col) %} {% endif %} {% endfor %} + {# -- Check for changed column types #} + {% for col in tmp_columns %} + {% for target_col in target_columns %} + {% if (target_col.column|lower) == (col.column|lower) %} + {% if target_col.dtype != col.dtype %} + {% do changed_cols.append({'column': col.column, 'from_type': target_col.dtype, 'to_type': col.dtype}) %} + {% break %} + {% endif %} + {% endif %} + {% endfor %} + {% endfor %} + {# -- Apply schema changes based on the on_schema_change setting #} {% if missing_cols and on_schema_change != 'ignore' %} {% if on_schema_change == 'fail' %} - {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ missing_cols | map(attribute='name') | join(', ') ~ '. Add columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} + {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ missing_cols | map(attribute='column') | join(', ') ~ '. Add columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} {% elif on_schema_change == 'append_new_columns' %} - {% do log("Adding new columns: " ~ missing_cols | map(attribute='name') | join(', '), info=True) %} + {% do log("Adding new columns: " ~ missing_cols | map(attribute='column') | join(', '), info=True) %} {% for col in missing_cols %} {% set sql_add_col %} - {{ doris_add_column(existing_relation, col.name, col.data_type) }} + {{ doris_add_column(existing_relation, col.column, col.data_type) }} {% endset %} {% do run_query(sql_add_col) %} {% endfor %} {% endif %} {% endif %} + + {% if changed_cols and on_schema_change != 'ignore' %} + {% if on_schema_change == 'fail' %} + {% do exceptions.raise_compiler_error('Schema changes detected: ' ~ changed_cols | map(attribute='column') | join(', ') ~ '. Alter columns to target or use "ignore" or "append_new_columns" as on_schema_change config.') %} + {% elif on_schema_change == 'append_new_columns' %} + {% do log("Alter columns: " ~ changed_cols | map(attribute='column') | join(', '), info=True) %} + {% for col in changed_cols %} + {% set sql_alter_col %} + {{ doris_alter_column_type(existing_relation, col.column, col.to_type) }} + {% endset %} + {% do run_query(sql_alter_col) %} + {% endfor %} + {% endif %} + {% endif %} {% endif %} {% endmacro %} \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..79d113f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,10 @@ +[project] +name = "dbt-doris" +version = "0.3.5" +description = "Add your description here" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + "dbt-doris>=0.3.4", + "mysql-connector-python>=9.3.0", +] diff --git a/setup.py b/setup.py index b16f13c..d2c0bd7 100644 --- a/setup.py +++ b/setup.py @@ -19,7 +19,7 @@ include_package_data=True, install_requires=[ f"dbt-core>={dbt_core_version}", - "mysql-connector-python>=8.0.0,<8.1", + "mysql-connector-python==9.3.0", "urllib3~=1.0", ], python_requires=">=3.7.2", diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..f868757 --- /dev/null +++ b/uv.lock @@ -0,0 +1,36 @@ +version = 1 +revision = 2 +requires-python = ">=3.12" + +[[package]] +name = "dbt-doris" +version = "0.3.5" +source = { virtual = "." } +dependencies = [ + { name = "mysql-connector-python" }, +] + +[package.metadata] +requires-dist = [ + { name = "dbt-doris", specifier = ">=0.3.4" }, + { name = "mysql-connector-python", specifier = ">=9.3.0" }, +] + +[[package]] +name = "mysql-connector-python" +version = "9.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/82/5e/55b265cb95938e271208e5692d7e615c53f2aeea894ab72a9f14ab198e9a/mysql-connector-python-9.3.0.tar.gz", hash = "sha256:8b16d51447e3603f18478fb5a19b333bfb73fb58f872eb055a105635f53d2345", size = 942579, upload-time = "2025-05-07T18:50:34.339Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bf/73/b42061ea4c0500edad4f92834ed7d75b1a740d11970e531c5be4dc1af5cd/mysql_connector_python-9.3.0-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:2589af070babdff9c920ee37f929218d80afa704f4e2a99f1ddcb13d19de4450", size = 15151288, upload-time = "2025-04-15T18:43:17.762Z" }, + { url = "https://files.pythonhosted.org/packages/27/87/9cd7e803c762c5098683c83837d2258c2f83cf82d33fabd1d0eaadae06ee/mysql_connector_python-9.3.0-cp312-cp312-macosx_14_0_x86_64.whl", hash = "sha256:1916256ecd039f4673715550d28138416bac5962335e06d36f7434c47feb5232", size = 15967397, upload-time = "2025-04-15T18:43:20.799Z" }, + { url = "https://files.pythonhosted.org/packages/5a/5d/cd63f31bf5d0536ee1e4216fb2f3f57175ca1e0dd37e1e8139083d2156e8/mysql_connector_python-9.3.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:d33e2f88e1d4b15844cfed2bb6e90612525ba2c1af2fb10b4a25b2c89a1fe49a", size = 33457025, upload-time = "2025-04-15T18:43:24.09Z" }, + { url = "https://files.pythonhosted.org/packages/76/65/9609a96edc0d015d1017176974c42b955cf87ba92cd31765f99cba835715/mysql_connector_python-9.3.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:0aedee809e1f8dbab6b2732f51ee1619b54a56d15b9070655bc31fb822c1a015", size = 33853427, upload-time = "2025-04-15T18:43:28.441Z" }, + { url = "https://files.pythonhosted.org/packages/c2/da/f81eeb5b63dea3ebe035fbbbdc036ae517155ad73f2e9640ee7c9eace09d/mysql_connector_python-9.3.0-cp312-cp312-win_amd64.whl", hash = "sha256:3853799f4b719357ea25eba05f5f278a158a85a5c8209b3d058947a948bc9262", size = 16358560, upload-time = "2025-04-15T18:43:32.281Z" }, + { url = "https://files.pythonhosted.org/packages/6a/16/5762061505a0d0d3a333613b6f5d7b8eb3222a689aa32f71ed15f1532ad1/mysql_connector_python-9.3.0-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:9516a4cdbaee3c9200f0e7d9aafb31057692f45c202cdcb43a3f9b37c94e7c84", size = 15151425, upload-time = "2025-04-15T18:43:35.573Z" }, + { url = "https://files.pythonhosted.org/packages/db/40/22de86e966e648ea0e3e438ad523c86d0cf4866b3841e248726fb4afded8/mysql_connector_python-9.3.0-cp313-cp313-macosx_14_0_x86_64.whl", hash = "sha256:495798dd34445d749991fb3a2aa87b4205100676939556d8d4aab5d5558e7a1f", size = 15967663, upload-time = "2025-04-15T18:43:38.248Z" }, + { url = "https://files.pythonhosted.org/packages/4c/19/36983937347b6a58af546950c88a9403cdce944893850e80ffb7f602a099/mysql_connector_python-9.3.0-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:be0ef15f6023ae2037347498f005a4471f694f8a6b8384c3194895e153120286", size = 33457288, upload-time = "2025-04-15T18:43:41.901Z" }, + { url = "https://files.pythonhosted.org/packages/18/12/7ccbc678a130df0f751596b37eddb98b2e40930d0ebc9ee41965ffbf0b92/mysql_connector_python-9.3.0-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:4364d3a37c449f1c0bb9e52fd4eddc620126b9897b6b9f2fd1b3f33dacc16356", size = 33853838, upload-time = "2025-04-15T18:43:45.505Z" }, + { url = "https://files.pythonhosted.org/packages/c2/5e/c361caa024ce14ffc1f5b153d90f0febf5e9483a60c4b5c84e1e012363cc/mysql_connector_python-9.3.0-cp313-cp313-win_amd64.whl", hash = "sha256:2a5de57814217077a8672063167b616b1034a37b614b93abcb602cc0b8c6fade", size = 16358561, upload-time = "2025-04-15T18:43:49.176Z" }, + { url = "https://files.pythonhosted.org/packages/23/1d/8c2c6672094b538f4881f7714e5332fdcddd05a7e196cbc9eb4a9b5e9a45/mysql_connector_python-9.3.0-py2.py3-none-any.whl", hash = "sha256:8ab7719d614cf5463521082fab86afc21ada504b538166090e00eeaa1ff729bc", size = 399302, upload-time = "2025-04-15T18:44:10.046Z" }, +] From 36bac579efd6b9efc8665fe160140d02ed26d4d7 Mon Sep 17 00:00:00 2001 From: "anh.tran" Date: Fri, 18 Sep 2026 22:26:11 +0700 Subject: [PATCH 15/15] Default to pure-Python connector to avoid 8KB statement truncation mysql-connector ships a C extension that is used by default whenever it is installed (`use_pure` defaults to False in mysql.connector.connect). That extension pushes statements through a fixed NET_BUFFER_LENGTH (8192) buffer, and Doris' FE does not reassemble the resulting split packets: anything longer than 8KB reaches the parser truncated mid-token. The failure is confusing because it surfaces as a syntax error pointing at whichever line happens to contain byte 8192, with a garbled token stream: no viable alternative at input 'CAST(JSONJSON(JSON'(line 155, pos 20) A model that flattens ~100 JSON columns crosses 8KB easily, so this is hit by any reasonably wide model rather than being an edge case. Add `use_pure` to DorisCredentials, defaulting to True, and pass it through to mysql.connector.connect(). Set `use_pure: false` in profiles.yml to opt back into the C extension if every statement stays under 8KB. Co-Authored-By: Claude Opus 5 (1M context) --- dbt/adapters/doris/connections.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/dbt/adapters/doris/connections.py b/dbt/adapters/doris/connections.py index 6585ea5..bf2b73e 100644 --- a/dbt/adapters/doris/connections.py +++ b/dbt/adapters/doris/connections.py @@ -41,6 +41,16 @@ class DorisCredentials(Credentials): password: str = "" database: Optional[str] = None schema: Optional[str] = None + # Use mysql-connector's pure-Python implementation. + # + # Defaults to True because the bundled C extension sends statements through a + # fixed NET_BUFFER_LENGTH (8192) buffer, and Doris' FE does not reassemble the + # split packets: any statement larger than 8KB arrives truncated mid-token and + # fails to parse, e.g. + # no viable alternative at input 'CAST(JSONJSON(JSON'(line 155, pos 20) + # Wide models (many columns / long CTEs) cross 8KB easily. Set this to false to + # opt back into the faster C extension when every statement stays under 8KB. + use_pure: bool = True @property @@ -48,7 +58,7 @@ def type(self): return "doris" def _connection_keys(self): - return "host", "port", "user", "schema" + return "host", "port", "user", "schema", "use_pure" @property def unique_field(self) -> str: @@ -81,6 +91,7 @@ def open(cls, connection: Connection) -> Connection: "buffered": True, "charset": "utf8", "get_warnings": True, + "use_pure": credentials.use_pure, } try: