fix(doris): re-raise genuine column DDL failures instead of swallowing them

pyproject.toml

@@ -152,7 +152,7 @@ build-test = [
 ]
 format = ["ruff"]
 type-stubs = ["types-psutil", "asyncpg-stubs"]
-ci-enabled-optional-deps = ["pydantic>=2.11.9", "asyncpg>=0.31.0", "neo4j>=5.18.0"]
+ci-enabled-optional-deps = ["pydantic>=2.11.9", "asyncpg>=0.31.0", "neo4j>=5.18.0", "pymysql>=1.1.0", "aiohttp>=3.9.0"]
 examples = ["sentence-transformers>=3.0.0", "numpy"]
 
 ci = [

python/cocoindex/connectors/doris/_target.py

@@ -1031,6 +1031,73 @@ class _TableAction(NamedTuple):
     column_actions: dict[str, statediff.DiffAction]
 
 
+def _is_benign_column_ddl_error(e: Exception, *, expect_exists: bool) -> bool:
+    """True if `e` reports the column DDL is already a no-op for the
+    direction being applied (e.g. ADD COLUMN on a column that is already
+    there, or DROP COLUMN on one that is already gone): a benign race or
+    an idempotent re-run. False for a genuine failure, which the caller
+    must re-raise rather than silently swallow.
+    """
+    msg = str(e).lower()
+    if expect_exists:
+        return "already exist" in msg or "duplicate column" in msg
+    return "not exist" in msg or "unknown column" in msg
+
+
+def _apply_column_actions(
+    config: DorisConnectionConfig,
+    table_name: str,
+    table_schema: TableSchema[Any],
+    column_actions: dict[str, statediff.DiffAction],
+) -> None:
+    """Apply non-PK column ALTER TABLE actions for one table.
+
+    A benign no-op (per `_is_benign_column_ddl_error`) is ignored. Any other
+    DDL failure is logged and re-raised, so the caller's reconcile() does
+    not commit a tracking record for a schema change that never actually
+    happened: mirrors the sqlite connector's own handling of the same bug
+    class.
+    """
+    for sub_key, col_action in column_actions.items():
+        if not sub_key.startswith(_COL_SUBKEY_PREFIX):
+            continue
+        col_name = sub_key[len(_COL_SUBKEY_PREFIX) :]
+        if col_name in table_schema.primary_key:
+            continue
+
+        col_def = table_schema.columns.get(col_name)
+        if col_action == "delete":
+            ddl = f"ALTER TABLE `{config.database}`.`{table_name}` DROP COLUMN `{col_name}`"
+            try:
+                _execute_ddl_sync(config, ddl)
+            except Exception as e:
+                if not _is_benign_column_ddl_error(e, expect_exists=False):
+                    _logger.warning(
+                        "Failed to drop column %s from table %s: %s",
+                        col_name,
+                        table_name,
+                        e,
+                    )
+                    raise
+        elif col_action in ("insert", "upsert") and col_def is not None:
+            nullable = "NULL" if col_def.nullable else "NOT NULL"
+            ddl = (
+                f"ALTER TABLE `{config.database}`.`{table_name}` "
+                f"ADD COLUMN `{col_name}` {col_def.type} {nullable}"
+            )
+            try:
+                _execute_ddl_sync(config, ddl)
+            except Exception as e:
+                if not _is_benign_column_ddl_error(e, expect_exists=True):
+                    _logger.warning(
+                        "Failed to add column %s to table %s: %s",
+                        col_name,
+                        table_name,
+                        e,
+                    )
+                    raise
+
+
 def _apply_table_actions(
     context_provider: ContextProvider,
     actions: Sequence[_TableAction],
@@ -1093,32 +1160,12 @@ def _apply_table_actions(
 
             # Reconcile non-PK columns incrementally
             if action.column_actions:
-                for sub_key, col_action in action.column_actions.items():
-                    if not sub_key.startswith(_COL_SUBKEY_PREFIX):
-                        continue
-                    col_name = sub_key[len(_COL_SUBKEY_PREFIX) :]
-                    if col_name in spec.table_schema.primary_key:
-                        continue
-
-                    col_def = spec.table_schema.columns.get(col_name)
-                    if col_action == "delete":
-                        try:
-                            _execute_ddl_sync(
-                                config,
-                                f"ALTER TABLE `{config.database}`.`{key.table_name}` DROP COLUMN `{col_name}`",
-                            )
-                        except Exception:
-                            pass
-                    elif col_action in ("insert", "upsert") and col_def is not None:
-                        nullable = "NULL" if col_def.nullable else "NOT NULL"
-                        try:
-                            _execute_ddl_sync(
-                                config,
-                                f"ALTER TABLE `{config.database}`.`{key.table_name}` "
-                                f"ADD COLUMN `{col_name}` {col_def.type} {nullable}",
-                            )
-                        except Exception:
-                            pass
+                _apply_column_actions(
+                    config,
+                    key.table_name,
+                    spec.table_schema,
+                    action.column_actions,
+                )
 
     return outputs
 

python/tests/connectors/test_doris_target_ddl_swallow.py

@@ -0,0 +1,240 @@
+"""Regression tests for the Doris connector's column-DDL error handling.
+
+Unlike test_doris_target.py, these do not need a live Doris cluster: they
+call `_apply_column_actions` directly and stub `_execute_ddl_sync`, since the
+bug and its fix are both pure control flow (which exceptions from a DDL
+statement are swallowed vs. re-raised), not something a live cluster
+response is needed to observe.
+"""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+try:
+    import pymysql  # type: ignore[import-untyped]  # noqa: F401
+    import aiohttp  # type: ignore[import-untyped]  # noqa: F401
+
+    DEPS_AVAILABLE = True
+except ImportError:
+    DEPS_AVAILABLE = False
+
+pytestmark = pytest.mark.skipif(
+    not DEPS_AVAILABLE, reason="pymysql/aiohttp not installed"
+)
+
+if DEPS_AVAILABLE:
+    from cocoindex._internal.context_keys import ContextKey, ContextProvider
+    from cocoindex.connectorkits import target as connector_target
+    from cocoindex.connectors import doris
+    from cocoindex.connectors.doris import _target as doris_target
+    from cocoindex.connectors.doris._target import (
+        DorisConnectionConfig,
+        TableSchema,
+        _apply_table_actions,
+        _TableAction,
+        _TableKey,
+        _TableSpec,
+    )
+    # `_apply_column_actions` / `_is_benign_column_ddl_error` are new symbols
+    # this fix introduces; imported lazily inside the tests that need them so
+    # the rest of this module (incl. the end-to-end test below, which only
+    # names pre-existing symbols) still collects and runs against pre-fix
+    # code, and demonstrates the actual swallowed-exception behavior rather
+    # than just an ImportError.
+
+
+def _config() -> "DorisConnectionConfig":
+    return doris.DorisConnectionConfig(fe_host="doris-test-host", database="testdb")
+
+
+def _schema() -> "TableSchema[dict[str, object]]":
+    return doris.TableSchema(
+        columns={
+            "id": doris.ColumnDef(type="VARCHAR(64)", nullable=False),
+            "col1": doris.ColumnDef(type="VARCHAR(255)", nullable=True),
+        },
+        primary_key=["id"],
+    )
+
+
+# ============================================================
+# _is_benign_column_ddl_error
+# ============================================================
+
+
+def test_benign_add_column_errors_are_recognized() -> None:
+    from cocoindex.connectors.doris._target import _is_benign_column_ddl_error
+
+    assert _is_benign_column_ddl_error(
+        Exception("Duplicate column name 'col1'"), expect_exists=True
+    )
+    assert _is_benign_column_ddl_error(
+        Exception("errCode = 2, column col1 already exists"), expect_exists=True
+    )
+
+
+def test_benign_drop_column_errors_are_recognized() -> None:
+    from cocoindex.connectors.doris._target import _is_benign_column_ddl_error
+
+    assert _is_benign_column_ddl_error(
+        Exception("Unknown column 'col1' in 'field list'"), expect_exists=False
+    )
+    assert _is_benign_column_ddl_error(
+        Exception("errCode = 2, column col1 does not exist"), expect_exists=False
+    )
+
+
+def test_genuine_errors_are_not_benign_in_either_direction() -> None:
+    from cocoindex.connectors.doris._target import _is_benign_column_ddl_error
+
+    permission_error = Exception("Access denied for user 'ro'@'%' to database 'testdb'")
+    assert not _is_benign_column_ddl_error(permission_error, expect_exists=True)
+    assert not _is_benign_column_ddl_error(permission_error, expect_exists=False)
+
+    in_progress = Exception("schema change job on table t already in progress")
+    assert not _is_benign_column_ddl_error(in_progress, expect_exists=True)
+    assert not _is_benign_column_ddl_error(in_progress, expect_exists=False)
+
+
+# ============================================================
+# _apply_column_actions
+# ============================================================
+
+
+def test_add_column_benign_duplicate_is_swallowed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise Exception("Duplicate column name 'col1'")
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    # Should not raise: an ADD COLUMN that already exists is a benign no-op.
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    _apply_column_actions(_config(), "t1", _schema(), {"col:col1": "insert"})
+
+
+def test_drop_column_benign_missing_is_swallowed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise Exception("Unknown column 'col1' in 'field list'")
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    # Should not raise: dropping an already-gone column is a benign no-op.
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    _apply_column_actions(_config(), "t1", _schema(), {"col:col1": "delete"})
+
+
+def test_add_column_genuine_failure_propagates_and_is_logged(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise Exception("Access denied for user 'ro'@'%' to database 'testdb'")
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    with caplog.at_level(logging.WARNING, logger="cocoindex.connectors.doris._target"):
+        with pytest.raises(Exception, match="Access denied"):
+            _apply_column_actions(_config(), "t1", _schema(), {"col:col1": "insert"})
+    assert any("Failed to add column col1" in r.message for r in caplog.records)
+
+
+def test_drop_column_genuine_failure_propagates_and_is_logged(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise Exception("schema change job on table t1 already in progress")
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    with caplog.at_level(logging.WARNING, logger="cocoindex.connectors.doris._target"):
+        with pytest.raises(Exception, match="already in progress"):
+            _apply_column_actions(_config(), "t1", _schema(), {"col:col1": "delete"})
+    assert any("Failed to drop column col1" in r.message for r in caplog.records)
+
+
+def test_add_column_success_reaches_ddl_unchanged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls = []
+
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        calls.append(sql)
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    _apply_column_actions(_config(), "t1", _schema(), {"col:col1": "upsert"})
+    assert len(calls) == 1
+    assert "ADD COLUMN `col1`" in calls[0]
+
+
+def test_pk_columns_are_never_altered(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise AssertionError(f"should not run DDL for a PK column: {sql}")
+
+    monkeypatch.setattr(
+        "cocoindex.connectors.doris._target._execute_ddl_sync", fake_execute
+    )
+    from cocoindex.connectors.doris._target import _apply_column_actions
+
+    _apply_column_actions(_config(), "t1", _schema(), {"col:id": "delete"})
+
+
+# ============================================================
+# End-to-end through _apply_table_actions (the real entry point Doris'
+# TargetActionSink dispatches into, not just the extracted helper)
+# ============================================================
+
+
+def test_end_to_end_genuine_ddl_failure_propagates_through_apply_table_actions(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A genuine ALTER TABLE failure during column reconciliation must
+    surface out of `_apply_table_actions`, not be swallowed. On the
+    pre-fix code this call returns normally, letting the caller commit a
+    tracking record for a schema change that never actually happened.
+    """
+    spec = _TableSpec(
+        table_schema=_schema(), managed_by=connector_target.ManagedBy.SYSTEM
+    )
+    managed_conn = doris.ManagedConnection(config=_config())
+
+    db_key: ContextKey[doris.ManagedConnection] = ContextKey(
+        "test_doris_target_ddl_swallow_e2e_db_key"
+    )
+    ctx = ContextProvider()
+    ctx.provide(db_key, managed_conn)
+
+    table_key = _TableKey(db_key=db_key.key, table_name="t1")
+    action = _TableAction(
+        key=table_key,
+        spec=spec,
+        main_action=None,
+        column_actions={"col:col1": "insert"},
+    )
+
+    def fake_execute(config: "DorisConnectionConfig", sql: str) -> None:
+        raise Exception("Access denied for user 'ro'@'%' to database 'testdb'")
+
+    monkeypatch.setattr(doris_target, "_execute_ddl_sync", fake_execute)
+
+    with pytest.raises(Exception, match="Access denied"):
+        _apply_table_actions(ctx, [action])