diff --git a/driver/statement.cpp b/driver/statement.cpp index 84be955d..d6523be6 100755 --- a/driver/statement.cpp +++ b/driver/statement.cpp @@ -72,7 +72,38 @@ void Statement::executeQuery(std::unique_ptr && mutator) { *param_set_processed_ptr = 0; next_param_set_idx = 0; + + const auto param_set_array_size = + getEffectiveDescriptor(SQL_ATTR_APP_PARAM_DESC).getAttrAs(SQL_DESC_ARRAY_SIZE, 1); + + // A set the driver never reaches is SQL_PARAM_UNUSED; every set that is sent overwrites its + // own entry in requestNextPackOfResultSets. + if (auto * param_status_ptr = getEffectiveDescriptor(SQL_ATTR_IMP_PARAM_DESC) + .getAttrAs(SQL_DESC_ARRAY_STATUS_PTR, 0)) { + for (SQLULEN i = 0; i < param_set_array_size; ++i) + param_status_ptr[i] = SQL_PARAM_UNUSED; + } + requestNextPackOfResultSets(std::move(mutator)); + + // ODBC executes the statement once per parameter set during SQLExecute; SQLMoreResults then + // walks the result sets those executions produced. A statement that returns nothing produces + // none, so its remaining sets have to be sent here rather than left to SQLMoreResults. Stop + // as soon as a set produces a result set, so a result-returning statement keeps handing them + // back one per SQLMoreResults. + if (!hasResultSet()) { + while (next_param_set_idx < param_set_array_size) { + std::unique_ptr carried_mutator; + if (result_reader) + carried_mutator = result_reader->releaseMutator(); + + requestNextPackOfResultSets(std::move(carried_mutator)); + + if (hasResultSet()) + break; + } + } + is_executed = true; } @@ -122,6 +153,30 @@ void Statement::requestNextPackOfResultSets(std::unique_ptr && mu if (next_param_set_idx >= param_set_array_size) return; + const auto param_set_idx = next_param_set_idx; + auto & ipd_desc = getEffectiveDescriptor(SQL_ATTR_IMP_PARAM_DESC); + auto * param_status_ptr = ipd_desc.getAttrAs(SQL_DESC_ARRAY_STATUS_PTR, 0); + + // SQL_ATTR_PARAMS_PROCESSED_PTR counts the sets processed including one that fails, so it is + // written before the attempt rather than after it. + // TODO: set this only after this single query is fully fetched (when output parameter support is added) + if (auto * processed_ptr = ipd_desc.getAttrAs(SQL_DESC_ROWS_PROCESSED_PTR, 0)) + *processed_ptr = param_set_idx + 1; + + try { + sendParamSet(std::move(mutator)); + } + catch (...) { + if (param_status_ptr) + param_status_ptr[param_set_idx] = SQL_PARAM_ERROR; + throw; + } + + if (param_status_ptr) + param_status_ptr[param_set_idx] = SQL_PARAM_SUCCESS; +} + +void Statement::sendParamSet(std::unique_ptr && mutator) { getDiagHeader().setAttr(SQL_DIAG_ROW_COUNT, -1); auto & connection = getParent(); @@ -135,11 +190,6 @@ void Statement::requestNextPackOfResultSets(std::unique_ptr && mu uri.addQueryParameter(key, value); } - // TODO: set this only after this single query is fully fetched (when output parameter support is added) - auto * param_set_processed_ptr = getEffectiveDescriptor(SQL_ATTR_IMP_PARAM_DESC).getAttrAs(SQL_DESC_ROWS_PROCESSED_PTR, 0); - if (param_set_processed_ptr) - *param_set_processed_ptr = next_param_set_idx; - Poco::Net::HTTPRequest request; request.setMethod(Poco::Net::HTTPRequest::HTTP_POST); request.setVersion(Poco::Net::HTTPRequest::HTTP_1_1); @@ -489,8 +539,6 @@ std::vector Statement::getParamsBindingInfo(std::size_t param_ if (fully_bound_param_count > 0) param_bindings.reserve(fully_bound_param_count); - auto * array_status_ptr = ipd_desc.getAttrAs(SQL_DESC_ARRAY_STATUS_PTR, 0); - const auto bind_type = apd_desc.getAttrAs(SQL_DESC_BIND_TYPE, SQL_PARAM_BIND_TYPE_DEFAULT); const auto * bind_offset_ptr = apd_desc.getAttrAs(SQL_DESC_BIND_OFFSET_PTR, 0); const auto bind_offset = (bind_offset_ptr ? *bind_offset_ptr : 0); @@ -531,9 +579,6 @@ std::vector Statement::getParamsBindingInfo(std::size_t param_ param_bindings.emplace_back(binding_info); } - if (array_status_ptr) - array_status_ptr[param_set_idx] = SQL_PARAM_SUCCESS; // TODO: elaborate? - return param_bindings; } diff --git a/driver/statement.h b/driver/statement.h index 249dfcd2..5cf0fd47 100644 --- a/driver/statement.h +++ b/driver/statement.h @@ -99,6 +99,7 @@ class Statement private: void requestNextPackOfResultSets(std::unique_ptr && mutator); + void sendParamSet(std::unique_ptr && mutator); /// Drops the keep-alive HTTP connection unless the previous response body was /// fully and cleanly consumed. diff --git a/driver/test/statement_parameter_bindings_it.cpp b/driver/test/statement_parameter_bindings_it.cpp index 332f07d1..797dd2bf 100755 --- a/driver/test/statement_parameter_bindings_it.cpp +++ b/driver/test/statement_parameter_bindings_it.cpp @@ -562,3 +562,84 @@ TEST_P(StringParameterBindingTest, SpecialCharactersRoundTrip) STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); } + +TEST_F(StatementParameterBindingsTest, IntArrayInsertProcessesEverySet) { + auto create_query = fromUTF8( + "CREATE OR REPLACE TABLE param_array_insert (i Int32) engine MergeTree order by i"); + STMT_OK(SQLExecDirect(hstmt, ptcharCast(create_query.data()), SQL_NTS)); + STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); + + SQLINTEGER param[] = { 10, 20, 30, 40, 50 }; + SQLLEN param_ind[] = { 0, 0, 0, 0, 0 }; + SQLULEN params_processed = 0; + + auto insert_query = fromUTF8("INSERT INTO param_array_insert (i) VALUES (?)"); + STMT_OK(SQLSetStmtAttr(hstmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)lengthof(param), 0)); + STMT_OK(SQLSetStmtAttr(hstmt, SQL_ATTR_PARAMS_PROCESSED_PTR, ¶ms_processed, 0)); + STMT_OK(SQLPrepare(hstmt, ptcharCast(insert_query.data()), SQL_NTS)); + STMT_OK(SQLBindParameter( + hstmt, + 1, + SQL_PARAM_INPUT, + getCTypeFor>(), + SQL_INTEGER, + 0, + 0, + param, + 0, + param_ind)); + + // An INSERT returns no result sets, so nothing would prompt a caller to call SQLMoreResults: + // this one call has to process the whole array. + STMT_OK(SQLExecute(hstmt)); + ASSERT_EQ(params_processed, lengthof(param)); + STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); + + auto select_query = fromUTF8("SELECT i FROM param_array_insert ORDER BY i"); + STMT_OK(SQLExecDirect(hstmt, ptcharCast(select_query.data()), SQL_NTS)); + for (std::size_t i = 0; i < lengthof(param); ++i) { + SQLINTEGER value = 0; + STMT_OK(SQLFetch(hstmt)); + STMT_OK(SQLGetData(hstmt, 1, SQL_C_SLONG, &value, sizeof(value), nullptr)); + ASSERT_EQ(value, param[i]) << "row " << i; + } + ASSERT_EQ(SQLFetch(hstmt), SQL_NO_DATA); + STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); +} + +TEST_F(StatementParameterBindingsTest, IntArrayInsertReportsTheFailingSet) { + // copy_id is derived from id with accurateCast, so an id that does not fit Int16 makes the + // server reject that parameter set and nothing after it is sent. + auto create_query = fromUTF8( + "CREATE OR REPLACE TABLE param_array_error " + "(id Int32, value String, copy_id Int16 DEFAULT accurateCast(id, 'Int16')) " + "engine MergeTree order by id"); + STMT_OK(SQLExecDirect(hstmt, ptcharCast(create_query.data()), SQL_NTS)); + STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); + + SQLINTEGER ids[] = { 1, 2, 40000, 4, 5 }; + SQLLEN id_ind[] = { 0, 0, 0, 0, 0 }; + char values[][2] = { "a", "b", "c", "d", "e" }; + SQLLEN value_ind[] = { SQL_NTS, SQL_NTS, SQL_NTS, SQL_NTS, SQL_NTS }; + SQLULEN params_processed = 0; + SQLUSMALLINT param_status[lengthof(ids)] = {}; + + auto insert_query = fromUTF8("INSERT INTO param_array_error (id, value) VALUES (?, ?)"); + STMT_OK(SQLSetStmtAttr(hstmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)lengthof(ids), 0)); + STMT_OK(SQLSetStmtAttr(hstmt, SQL_ATTR_PARAMS_PROCESSED_PTR, ¶ms_processed, 0)); + STMT_OK(SQLSetStmtAttr(hstmt, SQL_ATTR_PARAM_STATUS_PTR, param_status, 0)); + STMT_OK(SQLPrepare(hstmt, ptcharCast(insert_query.data()), SQL_NTS)); + STMT_OK(SQLBindParameter(hstmt, 1, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_INTEGER, 0, 0, ids, 0, id_ind)); + STMT_OK(SQLBindParameter(hstmt, 2, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 255, 0, values, sizeof(values[0]), value_ind)); + + ASSERT_EQ(SQLExecute(hstmt), SQL_ERROR); + + // The count includes the set that failed, and the status array says which one. + ASSERT_EQ(params_processed, 3u); + ASSERT_EQ(param_status[0], SQL_PARAM_SUCCESS); + ASSERT_EQ(param_status[1], SQL_PARAM_SUCCESS); + ASSERT_EQ(param_status[2], SQL_PARAM_ERROR); + ASSERT_EQ(param_status[3], SQL_PARAM_UNUSED); + ASSERT_EQ(param_status[4], SQL_PARAM_UNUSED); + STMT_OK(SQLFreeStmt(hstmt, SQL_CLOSE)); +}