@@ -581,6 +581,29 @@ void EnsureNativeFetchBufferSize(std::vector<ElementType>& buffer, size_t requir
     ResizeNativeFetchBuffer(buffer, newSize, reservedBytes);
 }
 
+size_t ExtendNativeFetchBuffer(std::vector<char>& buffer, size_t currentSize, size_t appendSize,
+                               size_t& reservedBytes) {
+    const size_t requiredSize =
+        CheckedAddSize(currentSize, appendSize, "Native fetch buffer size is too large");
+    EnsureNativeFetchBufferSize(buffer, requiredSize, reservedBytes);
+    return requiredSize;
+}
+
+py::tuple TestLobBufferGrowth(size_t chunkCount, size_t chunkSize) {
+    std::vector<char> buffer;
+    size_t dataSize = 0;
+    size_t reservedBytes = 0;
+    size_t growthCount = 0;
+    for (size_t chunk = 0; chunk < chunkCount; ++chunk) {
+        const size_t previousAllocation = buffer.size();
+        dataSize = ExtendNativeFetchBuffer(buffer, dataSize, chunkSize, reservedBytes);
+        if (buffer.size() != previousAllocation) {
+            ++growthCount;
+        }
+    }
+    return py::make_tuple(growthCount, dataSize, buffer.size(), reservedBytes);
+}
+
 template <typename ElementType>
 std::unique_ptr<ElementType[]> AllocateArrowArray(size_t count, size_t& reservedBytes,
                                                   const char* errorMessage) {
@@ -2484,6 +2507,7 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle,
                               bool use_prepare,
                               const py::dict& encoding_settings) {
     PERF_TIMER("SQLExecute_wrap");
+    ValidateInputSizes(input_sizes.ptr());
     if (!statementHandle || !statementHandle->get()) {
         return SQL_INVALID_HANDLE;
     }
@@ -3306,9 +3330,14 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list&
 
 SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle, const std::u16string& query,
                               const py::list& columnwise_params,
-                              std::vector<ParamInfo>& paramInfos, size_t paramSetSize,
+                              std::vector<ParamInfo>& paramInfos, py::handle paramSetSizeArg,
                               const py::dict& encodingSettings) {
     PERF_TIMER("SQLExecuteMany_wrap");
+    const size_t paramSetSize = static_cast<size_t>(
+        ValidateNativeRowCountArgument(paramSetSizeArg, "Parameter set size", false));
+    if (!statementHandle || !statementHandle->get()) {
+        return SQL_INVALID_HANDLE;
+    }
     statementHandle->resultMetadata.clear();
     LOG("SQLExecuteMany: Starting batch execution - param_count=%zu, "
         "param_set_size=%zu",
@@ -3646,6 +3675,7 @@ static py::object FetchLobColumnDataImpl(
     SQLGetDataFunc getData, bool (*hasTruncationDiagnostic)(SQLHSTMT)) {
     PERF_TIMER("FetchLobColumnData");
     std::vector<char> buffer;
+    size_t dataSize = 0;
     size_t reservedBytes = 0;
     ReserveNativeFetchBytes(reservedBytes, DAE_CHUNK_SIZE, sizeof(char));
     SQLRETURN ret = SQL_SUCCESS_WITH_INFO;
@@ -3742,10 +3772,8 @@ static py::object FetchLobColumnDataImpl(
             ThrowStdException("LOB fetch truncation made no progress");
         }
         if (bytesRead > 0) {
-            const size_t previousSize = buffer.size();
-            const size_t requiredSize =
-                CheckedAddSize(previousSize, bytesRead, "LOB fetch buffer is too large");
-            ResizeNativeFetchBuffer(buffer, requiredSize, reservedBytes);
+            const size_t previousSize = dataSize;
+            dataSize = ExtendNativeFetchBuffer(buffer, dataSize, bytesRead, reservedBytes);
             std::copy_n(chunk.data(), bytesRead, buffer.data() + previousSize);
             LOG("FetchLobColumnData: Appended %zu bytes at loop %d", bytesRead, loopCount);
         }
@@ -3755,21 +3783,21 @@ static py::object FetchLobColumnDataImpl(
             break;
         }
     }
-    LOG("FetchLobColumnData: Total bytes collected=%zu for column %d", buffer.size(), colIndex);
+    LOG("FetchLobColumnData: Total bytes collected=%zu for column %d", dataSize, colIndex);
 
-    if (buffer.empty()) {
+    if (dataSize == 0) {
         if (isBinary) {
             return py::bytes("");
         }
         return py::str("");
     }
     if (isWideChar) {
-        if (buffer.size() % sizeof(SQLWCHAR) != 0) {
+        if (dataSize % sizeof(SQLWCHAR) != 0) {
             ThrowStdException("Wide-character LOB data has an invalid byte length");
         }
         int byteOrder = -1;
         PyObject* decoded =
-            PyUnicode_DecodeUTF16(buffer.data(), static_cast<Py_ssize_t>(buffer.size()),
+            PyUnicode_DecodeUTF16(buffer.data(), static_cast<Py_ssize_t>(dataSize),
                                   "strict", &byteOrder);
         if (decoded == nullptr) {
             throw py::error_already_set();
@@ -3779,18 +3807,18 @@ static py::object FetchLobColumnDataImpl(
     if (isBinary) {
         LOG("FetchLobColumnData: Returning binary data - %zu bytes for column "
             "%d",
-            buffer.size(), colIndex);
-        return py::bytes(buffer.data(), buffer.size());
+            dataSize, colIndex);
+        return py::bytes(buffer.data(), dataSize);
     }
 
     // For SQL_C_CHAR data, decode using the appropriate encoding.
     const std::string effectiveCharEncoding = GetEffectiveCharDecoding(charEncoding);
-    py::bytes raw_bytes(buffer.data(), buffer.size());
+    py::bytes raw_bytes(buffer.data(), dataSize);
     try {
         py::object decoded = raw_bytes.attr("decode")(effectiveCharEncoding, "strict");
         LOG("FetchLobColumnData: Decoded narrow string with '%s' - %zu bytes -> %zu chars for "
             "column %d",
-            effectiveCharEncoding.c_str(), buffer.size(), py::len(decoded), colIndex);
+            effectiveCharEncoding.c_str(), dataSize, py::len(decoded), colIndex);
         return decoded;
     } catch (const py::error_already_set& e) {
         LOG_ERROR("FetchLobColumnData: Failed to decode with '%s' for column %d: %s",
@@ -7538,6 +7566,8 @@ PYBIND11_MODULE(ddbc_bindings, m) {
           "Fetch an arrow batch of given length from the result set", py::arg("StatementHandle"),
           py::arg("capsules"), py::arg("arrowBatchSize"), py::arg("charCtype"),
           py::arg("messages") = py::none());
+    m.def("_test_lob_buffer_growth", &TestLobBufferGrowth, py::arg("chunk_count"),
+          py::arg("chunk_size"));
     m.def("DDBCSQLFreeHandle", &SQLFreeHandle_wrap, "Free a handle");
     m.def("DDBCSQLResetStmt", &SQLResetStmt_wrap,
           "Close cursor and unbind params without freeing HSTMT");
 
@@ -274,6 +274,89 @@ inline void NormalizeTimeParam(PyObject* params, Py_ssize_t index, SQLULEN& colu
     }
 }
 
+inline long long ValidatedInputSizeInteger(PyObject* value, const char* fieldName) {
+    if (!PyLong_Check(value) || PyBool_Check(value)) {
+        throw py::type_error(std::string(fieldName) + " must be an integer");
+    }
+    int overflow = 0;
+    const long long result = PyLong_AsLongLongAndOverflow(value, &overflow);
+    if ((result == -1 && PyErr_Occurred()) || overflow != 0) {
+        PyErr_Clear();
+        throw py::value_error(std::string(fieldName) + " is out of range");
+    }
+    return result;
+}
+
+inline void ValidateInputSizes(PyObject* inputSizes) {
+    if (inputSizes == Py_None) {
+        return;
+    }
+    if (!PyList_Check(inputSizes)) {
+        throw py::type_error("inputSizes must be None or a list");
+    }
+
+    const Py_ssize_t count = PyList_GET_SIZE(inputSizes);
+    for (Py_ssize_t index = 0; index < count; ++index) {
+        PyObject* entry = PyList_GET_ITEM(inputSizes, index);
+        if (!PyTuple_Check(entry) || PyTuple_GET_SIZE(entry) != 4) {
+            throw py::type_error("each inputSizes entry must be a four-item tuple");
+        }
+
+        const long long sqlType =
+            ValidatedInputSizeInteger(PyTuple_GET_ITEM(entry, 0), "SQL type");
+        const long long cType =
+            ValidatedInputSizeInteger(PyTuple_GET_ITEM(entry, 1), "C type");
+        PyObject* columnSize = PyTuple_GET_ITEM(entry, 2);
+        PyObject* decimalDigits = PyTuple_GET_ITEM(entry, 3);
+        if (!PyLong_Check(columnSize) || PyBool_Check(columnSize) ||
+            !PyLong_Check(decimalDigits) || PyBool_Check(decimalDigits)) {
+            throw py::type_error("column size and decimal digits must be integers");
+        }
+
+        if (sqlType < std::numeric_limits<SQLSMALLINT>::min() ||
+            sqlType > std::numeric_limits<SQLSMALLINT>::max() ||
+            cType < std::numeric_limits<SQLSMALLINT>::min() ||
+            cType > std::numeric_limits<SQLSMALLINT>::max()) {
+            throw py::value_error("SQL and C types must fit in SQLSMALLINT");
+        }
+        const bool isNumeric = sqlType == SQL_DECIMAL || sqlType == SQL_NUMERIC;
+        py::int_ zero(0);
+        const int negativeColumnSize =
+            PyObject_RichCompareBool(columnSize, zero.ptr(), Py_LT);
+        const int negativeDecimalDigits =
+            PyObject_RichCompareBool(decimalDigits, zero.ptr(), Py_LT);
+        if (negativeColumnSize == -1 || negativeDecimalDigits == -1) {
+            throw py::error_already_set();
+        }
+        if (negativeColumnSize == 1) {
+            throw py::value_error("column size must be non-negative");
+        }
+        if (negativeDecimalDigits == 1) {
+            throw py::value_error("decimal digits must be non-negative");
+        }
+        if (!isNumeric) {
+            const unsigned long long requestedSize = PyLong_AsUnsignedLongLong(columnSize);
+            if (requestedSize == static_cast<unsigned long long>(-1) && PyErr_Occurred()) {
+                PyErr_Clear();
+                throw py::value_error("column size is out of range");
+            }
+            if (requestedSize > std::numeric_limits<SQLULEN>::max()) {
+                throw py::value_error("column size is out of range");
+            }
+            const unsigned long long requestedDigits =
+                PyLong_AsUnsignedLongLong(decimalDigits);
+            if (requestedDigits == static_cast<unsigned long long>(-1) && PyErr_Occurred()) {
+                PyErr_Clear();
+                throw py::value_error("decimal digits are out of range");
+            }
+            if (requestedDigits >
+                static_cast<unsigned long long>(std::numeric_limits<SQLSMALLINT>::max())) {
+                throw py::value_error("decimal digits are out of range");
+            }
+        }
+    }
+}
+
 inline void ApplyInputSizeOverride(PyObject* params, PyObject* inputSize, Py_ssize_t index,
                                    ParamInfo& info, const std::string& charEncoding) {
     py::tuple values = borrow<py::tuple>(inputSize);
@@ -391,6 +474,11 @@ inline std::vector<ParamInfo> DetectParamTypes(PyObject* params, PyObject* input
                                                const std::string& charEncoding = "utf-8") {
     PyTypeCache::initialize();
 
+    if (!PyList_Check(params)) {
+        throw py::type_error("params must be a list");
+    }
+    ValidateInputSizes(inputSizes);
+
     const Py_ssize_t n = PyList_GET_SIZE(params);
     const Py_ssize_t inputSizeCount = inputSizes == Py_None ? 0 : PyList_GET_SIZE(inputSizes);
     std::vector<ParamInfo> infos(n);
 
@@ -139,6 +139,42 @@ def test_lob_unrelated_warning_with_progress_completes():
     assert calls == 1
 
 
+@pytest.mark.skipif(not DDBC_AVAILABLE, reason="ddbc_bindings not available")
+@pytest.mark.parametrize("param_set_size", [-1, 0, 1_000_001, True])
+def test_executemany_rejects_unsafe_size_before_handle_access(param_set_size):
+    with pytest.raises(RuntimeError, match="Parameter set size"):
+        ddbc.SQLExecuteMany(None, "", [], [], param_set_size, {})
+
+
+@pytest.mark.skipif(not DDBC_AVAILABLE, reason="ddbc_bindings not available")
+@pytest.mark.parametrize(
+    ("input_sizes", "message"),
+    [
+        ((12, -8, 1, 0), "inputSizes must be None or a list"),
+        ([None], "four-item tuple"),
+        ([(12, -8, 1)], "four-item tuple"),
+        ([(12, -8, True, 0)], "must be integers"),
+        ([(12, -8, -1, 0)], "column size must be non-negative"),
+        ([(12, -8, 2**128, 0)], "column size is out of range"),
+        ([(12, -8, 1, 2**128)], "decimal digits are out of range"),
+    ],
+)
+def test_execute_rejects_malformed_input_sizes_before_handle_access(input_sizes, message):
+    with pytest.raises((TypeError, ValueError), match=message):
+        ddbc.DDBCSQLExecute(None, "", [], input_sizes, [False], True, {})
+
+
+@pytest.mark.skipif(not DDBC_AVAILABLE, reason="ddbc_bindings not available")
+def test_lob_buffer_growth_is_geometric():
+    growth_count, data_size, allocation_size, reserved_bytes = ddbc._test_lob_buffer_growth(
+        64, 64 * 1024
+    )
+    assert data_size == 4 * 1024 * 1024
+    assert allocation_size == data_size
+    assert reserved_bytes == allocation_size
+    assert growth_count == 7
+
+
 @pytest.mark.skipif(not DDBC_AVAILABLE, reason="ddbc_bindings not available")
 class TestPybindModuleInfo:
     """Test module information and architecture detection."""