fix(datasource): filter Hive partition metadata from table fields

backend/apps/db/db.py

@@ -598,7 +598,19 @@ def get_fields(ds: CoreDatasource, table_name: str = None):
             with get_driver_pool(ds).connection() as conn, conn.cursor() as cursor:
                 cursor.execute(sql)
                 res = cursor.fetchall()
-                res_list = [ColumnSchema(*item[:3]) for item in res]
+                res_list = []
+                seen_fields = set()
+                for item in res:
+                    field_name = item[0]
+                    if not field_name or not field_name.strip():
+                        continue
+                    # DESCRIBE includes section headings and repeats partition columns.
+                    if field_name.lstrip().startswith('#') and (item[1] or '').strip() in ('', 'data_type'):
+                        continue
+                    if field_name in seen_fields:
+                        continue
+                    seen_fields.add(field_name)
+                    res_list.append(ColumnSchema(*item[:3]))
                 return res_list
 
 

backend/tests/test_hive_fields.py

@@ -0,0 +1,124 @@
+"""Regression tests for Hive DESCRIBE field discovery (issue #1250)."""
+
+import ast
+import json
+from contextlib import nullcontext
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+BACKEND_DIR = Path(__file__).resolve().parents[1]
+
+
+def _load_symbol(relative_path, name, namespace):
+    # As in test_connection_pool_manager, avoid importing unrelated DB drivers.
+    path = BACKEND_DIR / relative_path
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    node = next(node for node in tree.body if getattr(node, "name", None) == name)
+    module = ast.Module(body=[node], type_ignores=[])
+    exec(compile(module, str(path), "exec"), namespace)
+    return namespace[name]
+
+
+@pytest.fixture
+def get_hive_fields():
+    cursor = Mock()
+    connection = SimpleNamespace(cursor=lambda: nullcontext(cursor))
+    pool = SimpleNamespace(connection=lambda: nullcontext(connection))
+    namespace = {
+        "json": json,
+        "CoreDatasource": SimpleNamespace,
+        "DatasourceConf": SimpleNamespace,
+        "aes_decrypt": lambda value: value,
+        "DB": SimpleNamespace(
+            get_db=lambda _: SimpleNamespace(connect_type="driver")
+        ),
+        "ConnectType": SimpleNamespace(sqlalchemy="sqlalchemy"),
+        "get_driver_pool": lambda _: pool,
+        "get_field_sql": lambda *args: ("DESCRIBE sample", None, None),
+    }
+    _load_symbol("common/utils/utils.py", "equals_ignore_case", namespace)
+    _load_symbol("apps/datasource/models/datasource.py", "ColumnSchema", namespace)
+    get_fields = _load_symbol("apps/db/db.py", "get_fields", namespace)
+
+    def discover(rows):
+        cursor.fetchall.return_value = rows
+        fields = get_fields(SimpleNamespace(type="hive", configuration="{}"), "sample")
+        return [(field.fieldName, field.fieldType, field.fieldComment) for field in fields]
+
+    return discover
+
+
+@pytest.mark.parametrize("repeat_partition", [False, True])
+def test_partition_fields_are_retained_once_without_metadata(get_hive_fields, repeat_partition):
+    rows = [("id", "int", "identifier")]
+    if repeat_partition:
+        rows.append(("dt", "string", "partition date"))
+    rows.extend([
+        ("", "", ""),
+        ("# Partition Information", "", ""),
+        ("# col_name", "data_type", "comment"),
+        ("dt", "string", "partition date"),
+    ])
+
+    assert get_hive_fields(rows) == [
+        ("id", "int", "identifier"),
+        ("dt", "string", "partition date"),
+    ]
+
+
+def test_duplicate_fields_preserve_first_occurrence_and_order(get_hive_fields):
+    assert get_hive_fields([
+        ("id", "int", "identifier"),
+        ("dt", "string", "first comment"),
+        ("region", "string", "region"),
+        ("dt", "string", "repeated comment"),
+    ]) == [
+        ("id", "int", "identifier"),
+        ("dt", "string", "first comment"),
+        ("region", "string", "region"),
+    ]
+
+
+def test_blank_names_and_padded_metadata_are_ignored(get_hive_fields):
+    assert get_hive_fields([
+        (None, None, None),
+        ("  ", "", ""),
+        ("  # Partition Information  ", None, None),
+        ("  # col_name  ", "data_type  ", "comment"),
+        ("dt", "string", None),
+    ]) == [("dt", "string", None)]
+
+
+def test_regular_columns_preserve_names_types_and_comments(get_hive_fields):
+    assert get_hive_fields([
+        ("id", "bigint", None),
+        ("amount", "decimal(10,2)", b"amount"),
+        ("#tag", "string", "tag"),
+        ("# Partition Information", "string", "a real column"),
+        (" spaced name ", "string", "keep the original name"),
+    ]) == [
+        ("id", "bigint", None),
+        ("amount", "decimal(10,2)", "amount"),
+        ("#tag", "string", "tag"),
+        ("# Partition Information", "string", "a real column"),
+        (" spaced name ", "string", "keep the original name"),
+    ]
+
+
+def test_extra_driver_columns_remain_supported(get_hive_fields):
+    assert get_hive_fields([
+        ("id", "int", "identifier", "extra", 1, None),
+        ("# Partition Information", None, None, "extra", 2, None),
+        ("dt", "string", "partition date", "extra", 3, None),
+    ]) == [
+        ("id", "int", "identifier"),
+        ("dt", "string", "partition date"),
+    ]
+
+
+@pytest.mark.parametrize("rows", [[], [("", "", ""), ("# col_name", "data_type", "comment")]])
+def test_no_real_columns_returns_empty_list(get_hive_fields, rows):
+    assert get_hive_fields(rows) == []