Fix/use pure connector 8kb truncation

.python-version

@@ -0,0 +1 @@
+3.12

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`.
 

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"

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"<DorisColumn {self.name} ({self.data_type})>"

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")
 
@@ -41,22 +41,32 @@ 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
     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:
         return self.schema
 
     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"
@@ -81,6 +91,7 @@ def open(cls, connection: Connection) -> Connection:
             "buffered": True,
             "charset": "utf8",
             "get_warnings": True,
+            "use_pure": credentials.use_pure,
         }
 
         try:
@@ -132,12 +143,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):

dbt/adapters/doris/impl.py

@@ -38,18 +38,18 @@
 )
 
 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
 from dbt.adapters.doris.connections import DorisConnectionManager
 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}"
             )
 
@@ -202,6 +203,37 @@ 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."""
+        self.execute_macro(
+            "doris__alter_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."""
+        self.execute_macro(
+            "doris__add_column",
+            kwargs={"relation": relation, "column_name": column_name, "column_type": column_type}
+        )
+
+    def drop_column(self, relation, column_name):
+        """Drop column from Doris table."""
+        self.execute_macro(
+            "doris__drop_column",
+            kwargs={"relation": relation, "column_name": column_name}
+        )
+
+    def rename_column(self, relation, old_column_name, new_column_name):
+        """Rename column in Doris table."""
+        self.execute_macro(
+            "doris__rename_column",
+            kwargs={
+                "relation": relation,
+                "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:

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

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"]

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']) -%}
@@ -38,3 +34,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 %}

dbt/include/doris/macros/adapters/relation.sql

@@ -24,11 +24,13 @@
 {% 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 PARTITION BY {% else %} PARTITION BY {% endif %}
+  {{ partition_type }} (
       {% for col in cols %}
         {{ col }}{% if not loop.last %},{% endif %}
       {% endfor %}