FIX: Request SQL_CHAR as SQL_C_WCHAR in arrow fetch path

mssql_python/cursor.py

@@ -2654,7 +2654,11 @@ def arrow_batch(self, batch_size: int = 8192) -> "pyarrow.RecordBatch":
  self._reset_rownumber()
 
  capsules = []
- ret = ddbc_bindings.DDBCSQLFetchArrowBatch(self.hstmt, capsules, max(batch_size, 0))
+ char_decoding = self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value)
+ char_c_type = char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value)
+ ret = ddbc_bindings.DDBCSQLFetchArrowBatch(
+ self.hstmt, capsules, max(batch_size, 0), char_c_type
+ )
  check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
 
  batch = pyarrow.RecordBatch._import_from_c_capsule(*capsules)

mssql_python/pybind/ddbc_bindings.cpp

@@ -4628,7 +4628,11 @@ int32_t days_from_civil(int y, int m, int d) {
 }
 
 SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
- int arrowBatchSize) {
+ int arrowBatchSize,
+ int charCtype) {
+ // Fetch narrow char data as SQL_C_CHAR if on Linux/macOS and configured by the user
+ charCtype = EffectiveCharCtypeForFetch(charCtype, "utf-8");
+
  // An overly large fetch size doesn't seem to help performance
  int fetchSize = 64;
 
@@ -4817,9 +4821,7 @@ SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
  ColumnBuffers buffers(numCols, fetchSize);
 
  if (!hasLobColumns && fetchSize > 0) {
- // Bind columns — Arrow always uses SQL_C_CHAR for VARCHAR because
- // it processes raw byte buffers directly, not via Python codecs.
- ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, SQL_C_CHAR);
+ ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype);
  if (!SQL_SUCCEEDED(ret)) {
  LOG("Error when binding columns");
  return ret;
@@ -4879,14 +4881,17 @@ SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
  case SQL_CHAR:
  case SQL_VARCHAR:
  case SQL_LONGVARCHAR: {
- ret = GetDataVar(hStmt, idxCol + 1, SQL_C_CHAR,
- buffers.charBuffers[idxCol],
- buffers.indicators[idxCol].data());
- if (!SQL_SUCCEEDED(ret)) {
- LOG("Error fetching CHAR LOB for column %d", idxCol + 1);
- return ret;
+ if (charCtype == SQL_C_CHAR) {
+ ret = GetDataVar(hStmt, idxCol + 1, SQL_C_CHAR,
+ buffers.charBuffers[idxCol],
+ buffers.indicators[idxCol].data());
+ if (!SQL_SUCCEEDED(ret)) {
+ LOG("Error fetching CHAR LOB data for column %d", idxCol + 1);
+ return ret;
+ }
+ break;
  }
-break;
+// else fall through to SQL_C_WCHAR case
  }
  case SQL_SS_XML:
  case SQL_WCHAR:
@@ -5131,27 +5136,31 @@ SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
  case SQL_CHAR:
  case SQL_VARCHAR:
  case SQL_LONGVARCHAR: {
+ if (charCtype == SQL_C_CHAR) {
 #if defined(__APPLE__) || defined(__linux__)
- uint64_t fetchBufferSize = columnSize * 4 + 1 /*null-terminator*/;
+uint64_t fetchBufferSize = columnSize * 4 + 1 /*null-terminator*/;
 #else
- uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
+uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
 #endif
- auto target_vec = &arrowColumnProducer->varData;
- auto start = arrowColumnProducer->varVal[idxRowArrow];
- while (target_vec->size() < start + dataLen) {
- target_vec->resize(target_vec->size() * 2);
- }
+auto target_vec = &arrowColumnProducer->varData;
+auto start = arrowColumnProducer->varVal[idxRowArrow];
+while (target_vec->size() < start + dataLen) {
+target_vec->resize(target_vec->size() * 2);
+}
 
- std::memcpy(&(*target_vec)[start],
- &buffers.charBuffers[idxCol][idxRowSql * fetchBufferSize],
- dataLen);
- arrowColumnProducer->varVal[idxRowArrow + 1] = start + dataLen;
- break;
+ std::memcpy(&(*target_vec)[start],
+ &buffers.charBuffers[idxCol][idxRowSql * fetchBufferSize],
+ dataLen);
+ arrowColumnProducer->varVal[idxRowArrow + 1] = start + dataLen;
+ break;
+ }
+ // else fall through to SQL_C_WCHAR case
  }
  case SQL_SS_XML:
  case SQL_WCHAR:
  case SQL_WVARCHAR:
  case SQL_WLONGVARCHAR: {
+ // We have previously fetched these as WCHARs, even for SQL_CHAR types.
  assert(dataLen % sizeof(SQLWCHAR) == 0);
  auto dataLenW = dataLen / sizeof(SQLWCHAR);
  auto wcharSource =

tests/test_004_cursor_arrow.py

@@ -313,6 +313,83 @@ def test_arrow_long_string(cursor: mssql_python.Cursor):
  assert batch.column(0).to_pylist() == [long_string]
 
 
+@pytest.mark.parametrize("sql_type", ["char(32)", "varchar(32)"])
+@pytest.mark.parametrize("narrow", [True, False])
+def test_arrow_char_utf8_collation_unicode(
+ cursor: mssql_python.Cursor, sql_type: str, narrow: bool
+):
+ table = "#t_arrow_char_decode"
+ collation = "Latin1_General_100_CI_AS_SC_UTF8"
+ expected = [
+ "Grüße",
+ "你好😀",
+ "こんにちは",
+ "Привет",
+ "Hello 世界",
+ "😀😃😄😁",
+ "",
+ None,
+ ]
+ if narrow:
+ cursor.connection.setdecoding(mssql_python.SQL_CHAR, ctype=mssql_python.SQL_CHAR)
+
+ try:
+ cursor.execute(
+ f"create table {table} (id int primary key, v {sql_type} collate {collation})"
+ )
+ except Exception as exc:
+ pytest.skip(f"UTF-8 collation '{collation}' not supported: {exc}")
+
+ try:
+ for index, value in enumerate(expected, start=1):
+ cursor.execute(f"insert into {table} (id, v) values (?, ?)", index, value)
+ tbl = cursor.execute(f"select v from {table} order by id").arrow()
+ assert tbl.column(0).type.equals(pa.large_string())
+ for expected_val, actual_val in zip(expected, tbl.column(0).to_pylist(), strict=True):
+ if actual_val is not None:
+ actual_val = actual_val.strip()
+ assert expected_val == actual_val
+ finally:
+ cursor.connection.setdecoding(mssql_python.SQL_CHAR)
+ cursor.execute(f"drop table if exists {table}")
+
+
+@pytest.mark.parametrize("sql_type", ["char(32)", "varchar(32)", "text"])
+@pytest.mark.parametrize("narrow", [True, False])
+def test_arrow_char_cp1252_collation_unicode(
+ cursor: mssql_python.Cursor, sql_type: str, narrow: bool
+):
+ table = "#t_arrow_char_decode"
+ collation = "SQL_Latin1_General_CP1_CI_AS"
+ expected = [
+ "Grüße",
+ "café René!",
+ "naïve café",
+ "Español",
+ "Müller-Öztürk",
+ "Françoise",
+ "",
+ None,
+ ]
+ if narrow:
+ cursor.connection.setdecoding(mssql_python.SQL_CHAR, ctype=mssql_python.SQL_CHAR)
+
+ cursor.execute(f"create table {table} (id int primary key, v {sql_type} collate {collation})")
+
+ try:
+ for index, value in enumerate(expected, start=1):
+ cursor.execute(f"insert into {table} (id, v) values (?, ?)", index, value)
+ tbl = cursor.execute(f"select v from {table} order by id").arrow()
+ assert tbl.column(0).type.equals(pa.large_string())
+ for expected_val, actual_val in zip(expected, tbl.column(0).to_pylist(), strict=True):
+ if actual_val is not None:
+ actual_val = actual_val.strip()
+ assert expected_val == actual_val
+ finally:
+ cursor.connection.setdecoding(mssql_python.SQL_CHAR)
+ cursor.execute(f"drop table if exists {table}")
+
+
 def test_rownumber_arrow_batch_interleaved_fetchmany(cursor: mssql_python.Cursor):
  """Verify that arrow_batch and fetchmany can be interleaved
  on the same result set with correct rownumber tracking and values."""