@@ -7,11 +7,14 @@
 belongs to the separate LOB decoder and is not covered by this regression.
 """
 
+import os
+import subprocess
 import sys
+import textwrap
 
 import pytest
 
-from mssql_python import SQL_CHAR, SQL_WCHAR
+from mssql_python import SQL_CHAR, SQL_WCHAR, ddbc_bindings as ddbc
 
 
 @pytest.fixture(scope="module")
@@ -294,3 +297,122 @@ def getdata(handle, column, ctype, buffer, capacity, indicator):
         check=False,
     )
     assert result.returncode == 0, result.stdout + result.stderr
+
+
+def _run_fetch_script(conn_str, script, *arguments, timeout=45):
+    result = subprocess.run(
+        [sys.executable, "-c", script, *arguments],
+        env={**os.environ, "DB_CONNECTION_STRING": conn_str},
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+@pytest.mark.parametrize("size", (1, 2))
+def test_getdata_decode_error_keeps_preexisting_and_completed_cells(cursor, size):
+    cursor.execute("SELECT 7, CAST(0x00D8 AS NVARCHAR(1))")
+    assert ddbc.DDBCSQLFetch(cursor.hstmt) == 0
+    marker = object()
+    row = [marker]
+    if size == 1:
+        assert ddbc.DDBCSQLGetData(cursor.hstmt, size, row, "utf-16le", "utf-16le", -8) == 0
+    else:
+        with pytest.raises(UnicodeDecodeError):
+            ddbc.DDBCSQLGetData(cursor.hstmt, size, row, "utf-16le", "utf-16le", -8)
+    assert row == [marker, 7]
+    cursor.execute("SELECT 42")
+    assert cursor.fetchval() == 42
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Windows does not export the native driver function-pointer globals",
+)
+@pytest.mark.parametrize("payload", ("utf8", "empty", "null", "invalid", "warning"))
+def test_narrow_getdata_decoding_in_subprocess(conn_str, payload):
+    script = textwrap.dedent("""
+        import ctypes as c
+        import os
+        import sys
+        import mssql_python
+        from mssql_python import ddbc_bindings as ddbc
+
+        mode = sys.argv[1]
+        data = b"\\xff" if mode == "invalid" else (
+            b"" if mode == "empty" else "\\ufeffA\\x00\\U0001f600".encode("utf-8")
+        )
+        pointer, short, length = c.c_void_p, c.c_short, c.c_ssize_t
+        get_type = c.CFUNCTYPE(short, pointer, c.c_ushort, short, pointer, length, pointer)
+        diag_type = c.CFUNCTYPE(
+            short, short, pointer, short, pointer, pointer, pointer, short, pointer
+        )
+        library = c.CDLL(ddbc.module.__file__)
+        slot = pointer.in_dll(library, "SQLGetData_ptr")
+        diag_slot = pointer.in_dll(library, "SQLGetDiagRec_ptr")
+        calls, errors = [], []
+
+        @diag_type
+        def diagnostic(handle_type, handle, record, state, native, message, capacity, size):
+            try:
+                if record > 1:
+                    return 100
+                text = "getdata warning".encode("utf-16le")
+                assert capacity > len(text) // 2
+                c.memmove(state, "01000\\0".encode("utf-16le"), 12)
+                c.memmove(message, text + b"\\0\\0", len(text) + 2)
+                c.cast(native, c.POINTER(c.c_int))[0] = 0
+                c.cast(size, c.POINTER(short))[0] = len(text) // 2
+                return 0
+            except BaseException as error:
+                errors.append(type(error).__name__)
+                return -1
+
+        @get_type
+        def getdata(handle, column, ctype, buffer, capacity, indicator):
+            try:
+                assert ctype == 1  # SQL_C_CHAR
+                calls.append(column)
+                result = original(handle, column, ctype, buffer, capacity, indicator)
+                assert result == 0 and capacity > len(data)
+                c.memmove(buffer, data + b"\\0", len(data) + 1)
+                c.cast(indicator, c.POINTER(length))[0] = -1 if mode == "null" else len(data)
+                return 1 if mode == "warning" else result
+            except BaseException as error:
+                errors.append(type(error).__name__)
+                return -1
+
+        with mssql_python.connect(os.environ["DB_CONNECTION_STRING"]) as connection:
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT CAST('text' AS VARCHAR(30))")
+                saved, saved_diag = slot.value, diag_slot.value
+                assert saved and saved_diag
+                original = get_type(saved)
+                assert ddbc.DDBCSQLFetch(cursor.hstmt) == 0
+                marker = object()
+                row = [marker]
+                slot.value = c.cast(getdata, pointer).value
+                if mode == "warning":
+                    diag_slot.value = c.cast(diagnostic, pointer).value
+                try:
+                    ret = ddbc.DDBCSQLGetData(
+                        cursor.hstmt, 1, row, "utf-8", "utf-16le", 1, cursor.messages
+                    )
+                    assert ret == (1 if mode == "warning" else 0)
+                finally:
+                    slot.value, diag_slot.value = saved, saved_diag
+                expected = None if mode == "null" else (
+                    data if mode == "invalid" else data.decode("utf-8")
+                )
+                assert row == [marker, expected]
+                assert type(row[1]) is type(expected)
+                assert calls == [1] and not errors, (calls, errors)
+                assert cursor.messages == (
+                    [("[01000] (0)", "getdata warning")] if mode == "warning" else []
+                )
+                cursor.execute("SELECT 42")
+                assert cursor.fetchval() == 42
+        """)
+    _run_fetch_script(conn_str, script, payload)