FEAT: Arrow Support in Bulk Copy

[gargsaumya]

Review Context:

1. mssql-py-core/src/arrow_bulkcopy.rs - the type contract: build_column_plans → resolve_kind (one dispatch per column per batch) → extract_value per cell. Every unsupported (Arrow, SQL) pair fails fast before any row is written.
2. mssql-py-core/src/cursor.rs - the bulkcopy_arrow binding: GIL released for the transfer, re-acquired only to pump each batch; batch read/import errors are surfaced as a hard error.
3. Python side is a thin wrapper mirroring bulkcopy.

In scope / supported destinations: bit, all int/uint (range-checked narrowing; uint64→bigint overflow-checked), real/float, (n)char/(n)varchar/(n)text, binary/varbinary/image (incl. any-width fixed_size_binary), uniqueidentifier, date, datetime2, time, datetimeoffset, decimal/numeric, money/smallmoney, datetime/smalldatetime, xml, and json.

Testing: 37 Rust unit tests + 122 Arrow integration tests + a long-haul stress job wired into CI.

Description

This pull request introduces comprehensive support and test coverage for Arrow-based bulk copy operations in the mssql-py-core project, focusing on the integration of the Arrow library for high-performance data loading. It adds new pyarrow-based tests for various SQL types, updates pipeline templates for more flexible test selection and reporting, and ensures the Arrow dependency is included in both the Rust and Python environments.

Arrow Bulk Copy Feature and Test Coverage

• Added new test modules for Arrow-based bulk copy (cursor.bulkcopy_arrow) covering SQL types: BIGINT, BINARY/VARBINARY, BIT, and CHAR, including edge cases like null handling and type-specific constraints. [1] [2] [3] [4]
• Introduced the arrow crate as a dependency in mssql-py-core/Cargo.toml to enable Arrow integration in the Rust backend.
• Added the arrow_bulkcopy Rust module to the project structure.

Pipeline and Test Infrastructure Improvements

• Updated .pipeline/templates/test-longhaul-template.yml to parameterize test selection, results file, and test run title, allowing for more flexible and parallelized long-haul test execution. [1] [2] [3]
• Modified .pipeline/templates/validation-stages.yml to run Arrow-specific long-haul tests in parallel with tuple-based tests, optimizing CI wall-clock time.
• Ensured pyarrow is installed in both pipeline and local development scripts (dev/test-python.sh).

These changes collectively improve the reliability, flexibility, and coverage of Arrow-based data loading in the project, and enhance the CI pipeline for future development.

Related Issues

Fixes #37 and AB#46077

Checklist

• cargo bfmt passes
• cargo bclippy passes
• cargo btest passes
• New/changed functionality has tests
• Public API changes are documented

[github-actions]

📊 Code Coverage Report

🔥 Diff Coverage 92%

🎯 Overall Coverage 90.1%

📦 Project: mssql-tds + mssql-odbc + mssql-py-core ℹ️ Note: diff coverage is reported, not enforced.

---

Diff Coverage

Diff: main...HEAD, staged and unstaged changes

• mssql-py-core/src/arrow_bulkcopy.rs (93.1%): Missing lines 149-154,156-161,362-365,437,451-454,463-466,578-580,607-613,672,705-708,738,773-776,914,1050,1146,1156,1164,1229,1244,1259,1277,1317,1336,1351,1371,1388,1432,1451,1466,1495,1507,1522,1552,1563,1578,1589,1601
• mssql-py-core/src/cursor.rs (89.4%): Missing lines 573,577,581,639,643,685,1225-1226,1237,1242-1243,1248-1251,1259-1267,1277-1279,1321,1324

Summary

• Total: 1284 lines
• Missing: 99 lines
• Coverage: 92%

mssql-py-core/src/arrow_bulkcopy.rs

  145     mappings
  146         .iter()
  147         .map(|m| {
  148             let field = schema.fields().get(m.source_index).ok_or_else(|| {
! 149                 Error::UsageError(format!(
! 150                     "Arrow source column index {} out of bounds (schema has {} fields)",
! 151                     m.source_index,
! 152                     schema.fields().len()
! 153                 ))
! 154             })?;
  155             let dest_meta = dest.get(m.destination_index).ok_or_else(|| {
! 156                 Error::UsageError(format!(
! 157                     "Destination column index {} out of bounds (table has {} columns)",
! 158                     m.destination_index,
! 159                     dest.len()
! 160                 ))
! 161             })?;
  162             let kind = resolve_kind(field.data_type(), dest_meta).map_err(|e| {
  163                 Error::UsageError(format!(
  164                     "Cannot map Arrow column '{}' ({:?}) to SQL column '{}' ({:?}): {}",
  165                     field.name(),

---

  358 
  359         match self.kind {
  360             ColumnPlanKind::Null => {
  361                 if !dest.is_nullable {
! 362                     return Err(Error::UsageError(format!(
! 363                         "Cannot insert NULL value into non-nullable column '{}'",
! 364                         dest.column_name
! 365                     )));
  366                 }
  367                 Ok(ColumnValues::Null)
  368             }
  369             ColumnPlanKind::Bool => {

---

  433                     a.value(row_idx).to_vec()
  434                 } else if let Some(a) = arr.as_any().downcast_ref::<FixedSizeBinaryArray>() {
  435                     a.value(row_idx).to_vec()
  436                 } else {
! 437                     return Err(downcast_err::<BinaryArray>(arr));
  438                 };
  439                 Ok(ColumnValues::Bytes(bytes))
  440             }
  441             ColumnPlanKind::LargeBinary => {

---

  447                 let sql_days = days
  448                     .checked_add(EPOCH_DAYS_FROM_YEAR_ONE)
  449                     .ok_or_else(|| Error::UsageError("Date32 value overflow".into()))?;
  450                 if sql_days < 0 {
! 451                     return Err(Error::UsageError(format!(
! 452                         "Date {} is before 0001-01-01",
! 453                         days
! 454                     )));
  455                 }
  456                 Ok(ColumnValues::Date(SqlDate::create(sql_days as u32)?))
  457             }
  458             ColumnPlanKind::Date64 => {

---

  459                 let ms = downcast::<Date64Array>(arr)?.value(row_idx);
  460                 let days = ms.div_euclid(86_400_000) as i64;
  461                 let sql_days = (days as i64) + EPOCH_DAYS_FROM_YEAR_ONE as i64;
  462                 if !(0..=3_652_058).contains(&sql_days) {
! 463                     return Err(Error::UsageError(format!(
! 464                         "Date64 value {} ms out of SQL DATE range",
! 465                         ms
! 466                     )));
  467                 }
  468                 Ok(ColumnValues::Date(SqlDate::create(sql_days as u32)?))
  469             }
  470             ColumnPlanKind::TimestampDateTime2 { unit, scale } => {

---

  574                         }
  575                     }
  576                 }
  577                 if !(0..=65_535).contains(&r_days) {
! 578                     return Err(Error::UsageError(
! 579                         "Timestamp out of SMALLDATETIME range after minute rounding".into(),
! 580                     ));
  581                 }
  582                 Ok(ColumnValues::SmallDateTime(SqlSmallDateTime {
  583                     days: r_days as u16,
  584                     time: r_hour * 60 + r_min,

---

  603         .downcast_ref::<T>()
  604         .ok_or_else(|| downcast_err::<T>(arr))
  605 }
  606 
! 607 fn downcast_err<T>(arr: &dyn Array) -> Error {
! 608     Error::UsageError(format!(
! 609         "Arrow array downcast failed: expected {}, got {:?}",
! 610         std::any::type_name::<T>(),
! 611         arr.data_type()
! 612     ))
! 613 }
  614 
  615 /// Days from 1900-01-01 to 1970-01-01 (UNIX epoch). Legacy `datetime`/
  616 /// `smalldatetime` count days from 1900, while Arrow timestamps count from the
  617 /// epoch.

---

  668         Ok(a.value(row_idx))
  669     } else if let Some(a) = arr.as_any().downcast_ref::<LargeStringArray>() {
  670         Ok(a.value(row_idx))
  671     } else {
! 672         Err(downcast_err::<StringArray>(arr))
  673     }
  674 }
  675 
  676 /// Read a timestamp value as 100-ns ticks since UNIX epoch.

---

  701     let days_since_epoch = ticks_since_epoch.div_euclid(ticks_per_day);
  702     let intra_day_ticks = ticks_since_epoch.rem_euclid(ticks_per_day);
  703     let sql_days = days_since_epoch + EPOCH_DAYS_FROM_YEAR_ONE as i64;
  704     if !(0..=3_652_058).contains(&sql_days) {
! 705         return Err(Error::UsageError(format!(
! 706             "Timestamp out of SQL DATETIME2 range: days_since_year_one={}",
! 707             sql_days
! 708         )));
  709     }
  710     Ok(SqlDateTime2 {
  711         days: sql_days as u32,
  712         time: SqlTime {

---

  734             .value(row_idx)
  735             .div_euclid(100),
  736     };
  737     if ticks < 0 {
! 738         return Err(Error::UsageError("Time value is negative".into()));
  739     }
  740     Ok(ticks as u64)
  741 }

---

  769             }
  770             Ok(ColumnValues::Int(v as i32))
  771         }
  772         SqlDbType::BigInt => Ok(ColumnValues::BigInt(v)),
! 773         _ => Err(Error::UsageError(format!(
! 774             "Unsupported destination type {:?} for integer column '{}'",
! 775             dest.sql_type, dest.column_name
! 776         ))),
  777     }
  778 }
  779 
  780 /// Format a `decimal128` raw value at the given scale as a decimal string for

---

  910         let arr: ArrayRef = Arc::new(StringArray::from(vec![Some("hello")]));
  911         let plan = one_col_plan(DataType::Utf8, &dest);
  912         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  913             ColumnValues::String(s) => assert_eq!(s.to_utf8_string(), "hello"),
! 914             other => panic!("expected String, got {:?}", other),
  915         }
  916     }
  917 
  918     #[test]

---

  1046         let arr: ArrayRef = Arc::new(StringArray::from(vec![Some(guid)]));
  1047         let plan = one_col_plan(DataType::Utf8, &dest);
  1048         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1049             ColumnValues::Uuid(u) => assert_eq!(u.to_string(), guid),
! 1050             other => panic!("expected Uuid, got {:?}", other),
  1051         }
  1052     }
  1053 
  1054     #[test]

---

  1142             .extract_value(a.as_ref(), 0, &dv)
  1143             .unwrap()
  1144         {
  1145             ColumnValues::String(s) => assert_eq!(s.to_utf8_string(), "abc"),
! 1146             o => panic!("expected String, got {o:?}"),
  1147         }
  1148 
  1149         let la: ArrayRef = Arc::new(LargeStringArray::from(vec![Some("h\u{e9}llo")]));
  1150         let dn = meta("n", SqlDbType::NVarChar, true);

---

  1152             .extract_value(la.as_ref(), 0, &dn)
  1153             .unwrap()
  1154         {
  1155             ColumnValues::String(s) => assert_eq!(s.to_utf8_string(), "h\u{e9}llo"),
! 1156             o => panic!("expected String, got {o:?}"),
  1157         }
  1158         let dlv = meta("lv", SqlDbType::VarChar, true);
  1159         match one_col_plan(DataType::LargeUtf8, &dlv)
  1160             .extract_value(la.as_ref(), 0, &dlv)

---

  1160             .extract_value(la.as_ref(), 0, &dlv)
  1161             .unwrap()
  1162         {
  1163             ColumnValues::String(s) => assert_eq!(s.to_utf8_string(), "h\u{e9}llo"),
! 1164             o => panic!("expected String, got {o:?}"),
  1165         }
  1166     }
  1167 
  1168     #[test]

---

  1225         let arr: ArrayRef = Arc::new(Date32Array::from(vec![Some(0)]));
  1226         let plan = one_col_plan(DataType::Date32, &dest);
  1227         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1228             ColumnValues::Date(d) => assert_eq!(d.get_days(), 719_162),
! 1229             other => panic!("expected Date, got {:?}", other),
  1230         }
  1231     }
  1232 
  1233     #[test]

---

  1240                 assert_eq!(dt2.days, 719_162);
  1241                 assert_eq!(dt2.time.time_nanoseconds, 0);
  1242                 assert_eq!(dt2.time.scale, 6);
  1243             }
! 1244             other => panic!("expected DateTime2, got {:?}", other),
  1245         }
  1246     }
  1247 
  1248     #[test]

---

  1255             ColumnValues::DateTime2(dt2) => {
  1256                 // 1µs = 10 ticks at 100-ns precision
  1257                 assert_eq!(dt2.time.time_nanoseconds, 10);
  1258             }
! 1259             other => panic!("expected DateTime2, got {:?}", other),
  1260         }
  1261     }
  1262 
  1263     #[test]

---

  1273                 // 1969-12-31 23:59:59.9999998 -> day 719_161, 863_999_999_998 ticks.
  1274                 assert_eq!(dt2.days, 719_161);
  1275                 assert_eq!(dt2.time.time_nanoseconds, 863_999_999_998);
  1276             }
! 1277             other => panic!("expected DateTime2, got {:?}", other),
  1278         }
  1279     }
  1280 
  1281     #[test]

---

  1313             ColumnValues::DateTimeOffset(dto) => {
  1314                 assert_eq!(dto.datetime2.days, 719_162);
  1315                 assert_eq!(dto.offset, 0);
  1316             }
! 1317             other => panic!("expected DateTimeOffset, got {:?}", other),
  1318         }
  1319     }
  1320 
  1321     #[test]

---

  1332                 assert_eq!(d.precision, 10);
  1333                 assert_eq!(d.scale, 2);
  1334                 assert_eq!(d.to_string(), "123.45");
  1335             }
! 1336             other => panic!("expected Decimal, got {:?}", other),
  1337         }
  1338     }
  1339 
  1340     #[test]

---

  1347         );
  1348         let plan = one_col_plan(DataType::Decimal128(10, 2), &dest);
  1349         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1350             ColumnValues::Decimal(d) => assert_eq!(d.to_string(), "-1.00"),
! 1351             other => panic!("expected Decimal, got {:?}", other),
  1352         }
  1353     }
  1354 
  1355     #[test]

---

  1367             ColumnValues::Decimal(d) => {
  1368                 assert_eq!(d.scale, 4);
  1369                 assert_eq!(d.to_string(), "123.4500");
  1370             }
! 1371             other => panic!("expected Decimal, got {:?}", other),
  1372         }
  1373     }
  1374 
  1375     #[test]

---

  1384         );
  1385         let plan = one_col_plan(DataType::Decimal128(10, 2), &dest);
  1386         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1387             ColumnValues::Numeric(d) => assert_eq!(d.to_string(), "123.45"),
! 1388             other => panic!("expected Numeric, got {:?}", other),
  1389         }
  1390     }
  1391 
  1392     #[test]

---

  1428             ColumnValues::Money(m) => {
  1429                 let money_val = ((m.lsb_part as i64) & 0xFFFF_FFFF) | ((m.msb_part as i64) << 32);
  1430                 assert_eq!(money_val, 1_234_500);
  1431             }
! 1432             other => panic!("expected Money, got {:?}", other),
  1433         }
  1434     }
  1435 
  1436     #[test]

---

  1447             ColumnValues::Money(m) => {
  1448                 let money_val = ((m.lsb_part as i64) & 0xFFFF_FFFF) | ((m.msb_part as i64) << 32);
  1449                 assert_eq!(money_val, -12346);
  1450             }
! 1451             other => panic!("expected Money, got {:?}", other),
  1452         }
  1453     }
  1454 
  1455     #[test]

---

  1462         );
  1463         let plan = one_col_plan(DataType::Decimal128(10, 2), &dest);
  1464         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1465             ColumnValues::SmallMoney(m) => assert_eq!(m.int_val, 1_234_500),
! 1466             other => panic!("expected SmallMoney, got {:?}", other),
  1467         }
  1468     }
  1469 
  1470     #[test]

---

  1491             ColumnValues::DateTime(dt) => {
  1492                 assert_eq!(dt.days, 25_567);
  1493                 assert_eq!(dt.time, 0);
  1494             }
! 1495             other => panic!("expected DateTime, got {:?}", other),
  1496         }
  1497     }
  1498 
  1499     #[test]

---

  1503         let arr: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![Some(1_000_000_i64)]));
  1504         let plan = one_col_plan(DataType::Timestamp(TimeUnit::Microsecond, None), &dest);
  1505         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1506             ColumnValues::DateTime(dt) => assert_eq!(dt.time, 300),
! 1507             other => panic!("expected DateTime, got {:?}", other),
  1508         }
  1509     }
  1510 
  1511     #[test]

---

  1518             ColumnValues::SmallDateTime(sdt) => {
  1519                 assert_eq!(sdt.days, 25_567);
  1520                 assert_eq!(sdt.time, 1);
  1521             }
! 1522             other => panic!("expected SmallDateTime, got {:?}", other),
  1523         }
  1524     }
  1525 
  1526     #[test]

---

  1548         let arr: ArrayRef = Arc::new(StringArray::from(vec![Some("<r/>")]));
  1549         let plan = one_col_plan(DataType::Utf8, &dest);
  1550         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1551             ColumnValues::Xml(x) => assert_eq!(x.as_string(), "<r/>"),
! 1552             other => panic!("expected Xml, got {:?}", other),
  1553         }
  1554     }
  1555 
  1556     #[test]

---

  1559         let arr: ArrayRef = Arc::new(StringArray::from(vec![Some("{\"a\":1}")]));
  1560         let plan = one_col_plan(DataType::Utf8, &dest);
  1561         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1562             ColumnValues::Json(j) => assert_eq!(j.bytes, b"{\"a\":1}"),
! 1563             other => panic!("expected Json, got {:?}", other),
  1564         }
  1565     }
  1566 
  1567     #[test]

---

  1574         );
  1575         let plan = one_col_plan(DataType::FixedSizeBinary(4), &dest);
  1576         match plan.extract_value(arr.as_ref(), 0, &dest).unwrap() {
  1577             ColumnValues::Bytes(b) => assert_eq!(b, vec![1, 2, 3, 4]),
! 1578             other => panic!("expected Bytes, got {:?}", other),
  1579         }
  1580     }
  1581 
  1582     #[test]

---

  1585         let dest = meta_dt2("ts", 0);
  1586         let plan = one_col_plan(DataType::Timestamp(TimeUnit::Microsecond, None), &dest);
  1587         match plan.kind {
  1588             ColumnPlanKind::TimestampDateTime2 { scale, .. } => assert_eq!(scale, 0),
! 1589             other => panic!("expected TimestampDateTime2, got {:?}", other),
  1590         }
  1591     }
  1592 
  1593     #[test]

---

  1597         dest.scale = 0;
  1598         let plan = one_col_plan(DataType::Time64(TimeUnit::Microsecond), &dest);
  1599         match plan.kind {
  1600             ColumnPlanKind::Time { scale, .. } => assert_eq!(scale, 0),
! 1601             other => panic!("expected Time, got {:?}", other),
  1602         }
  1603     }
  1604 }

---

mssql-py-core/src/cursor.rs

  569                 let destination_metadata = bulk_copy
  570                     .retrieve_destination_metadata()
  571                     .await
  572                     .map_err(|e| {
! 573                         error!(
  574                             "bulkcopy_arrow: Failed to retrieve destination metadata: {}",
  575                             e
  576                         );
! 577                         pyo3::exceptions::PyRuntimeError::new_err(format!(
  578                             "Failed to retrieve destination metadata: {}",
  579                             e
  580                         ))
! 581                     })?;
  582 
  583                 let mappings = if auto_generate_mappings {
  584                     let n = std::cmp::min(schema.fields().len(), destination_metadata.len());
  585                     (0..n)

---

  635                     bulk_copy = bulk_copy.add_column_mapping(tds_mapping);
  636                 }
  637 
  638                 let resolved_mappings = bulk_copy.get_resolved_mappings().await.map_err(|e| {
! 639                     pyo3::exceptions::PyRuntimeError::new_err(format!(
  640                         "Failed to resolve column mappings: {}",
  641                         e
  642                     ))
! 643                 })?;
  644 
  645                 let plans = build_column_plans(&schema, &destination_metadata, &resolved_mappings)
  646                     .map_err(convert_tds_error)?;
  647                 let plans_arc = Arc::new(plans);

---

  681         py_result.set_item("elapsed_time", result.elapsed.as_secs_f64())?;
  682         let rows_per_second = if result.elapsed.as_secs_f64() > 0.0 {
  683             result.rows_affected as f64 / result.elapsed.as_secs_f64()
  684         } else {
! 685             0.0
  686         };
  687         py_result.set_item("rows_per_second", rows_per_second)?;
  688         Ok(py_result.into())
  689     }

---

  1221         use pyo3::exceptions::{PyImportError, PyTypeError};
  1222         let py = source.py();
  1223 
  1224         let pyarrow = py.import("pyarrow").map_err(|e| {
! 1225             PyImportError::new_err(format!("pyarrow is required for bulkcopy_arrow: {}", e))
! 1226         })?;
  1227         let rbr_cls = pyarrow.getattr("RecordBatchReader")?;
  1228 
  1229         if source.hasattr("__arrow_c_stream__")? && rbr_cls.hasattr("from_stream")? {
  1230             // Covers pyarrow.Table, pandas.DataFrame (≥2.2), polars.DataFrame,

---

  1233             return Ok(reader.unbind());
  1234         }
  1235 
  1236         if source.is_instance(&rbr_cls)? {
! 1237             return Ok(source.clone().unbind());
  1238         }
  1239 
  1240         let table_cls = pyarrow.getattr("Table")?;
  1241         if source.is_instance(&table_cls)? {
! 1242             let reader = source.call_method0("to_reader")?;
! 1243             return Ok(reader.unbind());
  1244         }
  1245 
  1246         let batch_cls = pyarrow.getattr("RecordBatch")?;
  1247         if source.is_instance(&batch_cls)? {
! 1248             let schema = source.getattr("schema")?;
! 1249             let batches = pyo3::types::PyList::new(py, [source])?;
! 1250             let reader = rbr_cls.call_method1("from_batches", (schema, batches))?;
! 1251             return Ok(reader.unbind());
  1252         }
  1253 
  1254         // B1: single-batch producers exposing the Arrow PyCapsule *array*
  1255         // interface (`__arrow_c_array__`) but not the stream interface. Import

---

  1255         // interface (`__arrow_c_array__`) but not the stream interface. Import
  1256         // the one batch via the C data interface and wrap it as a single-batch
  1257         // reader. Stream producers were already handled above and take priority.
  1258         if source.hasattr("__arrow_c_array__")? {
! 1259             let capsules = source.call_method0("__arrow_c_array__")?;
! 1260             let schema_capsule = capsules.get_item(0)?;
! 1261             let array_capsule = capsules.get_item(1)?;
! 1262             let batch = batch_cls
! 1263                 .call_method1("_import_from_c_capsule", (schema_capsule, array_capsule))?;
! 1264             let schema = batch.getattr("schema")?;
! 1265             let batches = pyo3::types::PyList::new(py, [batch])?;
! 1266             let reader = rbr_cls.call_method1("from_batches", (schema, batches))?;
! 1267             return Ok(reader.unbind());
  1268         }
  1269 
  1270         // Last-resort: arbitrary iterable. Peek the first batch to learn the
  1271         // schema, then chain it back with the rest via from_batches.

---

  1273             let mut iter = iter;
  1274             let peeked = if let Some(item) = iter.by_ref().next() {
  1275                 let item = item?;
  1276                 if !item.is_instance(&batch_cls)? {
! 1277                     return Err(PyTypeError::new_err(
! 1278                         "iterable must yield pyarrow.RecordBatch instances",
! 1279                     ));
  1280                 }
  1281                 Some(item.unbind())
  1282             } else {
  1283                 None

---

  1317         let mut ffi_schema = FFI_ArrowSchema::empty();
  1318         let ptr_int = (&mut ffi_schema as *mut FFI_ArrowSchema) as usize;
  1319         schema_obj.call_method1("_export_to_c", (ptr_int,))?;
  1320         let schema = Schema::try_from(&ffi_schema).map_err(|e| {
! 1321             pyo3::exceptions::PyValueError::new_err(format!(
  1322                 "Failed to import Arrow schema via C data interface: {e}"
  1323             ))
! 1324         })?;
  1325         Ok(Arc::new(schema))
  1326     })
  1327 }

---

---

🔗 Quick Links

View Azure DevOps Build · Coverage Report

[Copilot]

Pull request overview

Adds an Arrow-based bulk-copy ingestion path to mssql-py-core so Python callers can bulk insert from pyarrow sources via Arrow’s C data interface, alongside a large integration-test suite and CI wiring (including a dedicated Arrow long-haul job).

Changes:

• Introduces PyCoreCursor.bulkcopy_arrow(...), Arrow source coercion, and an Arrow-to-TDS row adapter (arrow_bulkcopy).
• Adds broad integration coverage for Arrow → SQL type mappings, options, and column mapping behavior (plus a new Arrow long-haul test).
• Updates Python test runners / pipelines to install pyarrow and split Arrow long-haul execution into its own parallel job.

Reviewed changes

Copilot reviewed 34 out of 34 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
mssql-py-core/src/lib.rs	Registers the new arrow_bulkcopy module in the crate.
mssql-py-core/src/cursor.rs	Adds bulkcopy_arrow, Arrow source resolution, schema import, and Arrow batch iteration feeding the bulk-copy writer.
mssql-py-core/src/arrow_bulkcopy.rs	Implements Arrow type planning + per-row adaptation into BulkLoadRow for zero-copy-ish bulk-copy writes.
mssql-py-core/Cargo.toml	Adds the arrow dependency (FFI enabled) needed for Arrow C data interface interop.
dev/test-python.sh	Installs pyarrow in the Python test environment.
.pipeline/templates/validation-stages.yml	Adds a dedicated Arrow long-haul CI job and adjusts selectors to keep tuple/Arrow runs parallel.
.pipeline/templates/test-longhaul-template.yml	Parameterizes long-haul pytest selection/results output and installs pyarrow.
mssql-py-core/tests/test_bulkcopy_arrow_bigint.py	Integration coverage for BIGINT + uint64 overflow checks.
mssql-py-core/tests/test_bulkcopy_arrow_binary.py	Integration coverage for BINARY/VARBINARY Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_bit.py	Integration coverage for BIT Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_char.py	Integration coverage for CHAR Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_column_mismatch.py	Column mapping + count mismatch parity tests for the Arrow bulk copy path.
mssql-py-core/tests/test_bulkcopy_arrow_date.py	Integration coverage for DATE Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_datetime2.py	Integration coverage for DATETIME2 Arrow ingestion + tz-aware rejection behavior.
mssql-py-core/tests/test_bulkcopy_arrow_datetimeoffset.py	Integration coverage for DATETIMEOFFSET Arrow ingestion + UTC normalization behavior.
mssql-py-core/tests/test_bulkcopy_arrow_decimal.py	Integration coverage for DECIMAL Arrow ingestion, signs, and overflow scenarios.
mssql-py-core/tests/test_bulkcopy_arrow_float.py	Integration coverage for REAL/FLOAT Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_image.py	Integration coverage for IMAGE Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_int.py	Integration coverage for INT Arrow ingestion, int64 narrowing, and overflow checks.
mssql-py-core/tests/test_bulkcopy_arrow_longhaul.py	Adds an Arrow streaming long-haul stress test over a wide schema.
mssql-py-core/tests/test_bulkcopy_arrow_multipart_table_names.py	Arrow path parity tests for multipart/bracketed table-name handling.
mssql-py-core/tests/test_bulkcopy_arrow_nchar.py	Integration coverage for NCHAR Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_ntext.py	Integration coverage for NTEXT Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_numeric.py	Integration coverage for NUMERIC Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_nvarchar.py	Integration coverage for NVARCHAR Arrow ingestion incl. Unicode/MAX cases.
mssql-py-core/tests/test_bulkcopy_arrow_options.py	Integration coverage for bulk-copy options (triggers/constraints/nulls/identity/locks) via Arrow sources.
mssql-py-core/tests/test_bulkcopy_arrow_smallint.py	Integration coverage for SMALLINT Arrow ingestion and out-of-range handling.
mssql-py-core/tests/test_bulkcopy_arrow_text.py	Integration coverage for TEXT Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_time.py	Integration coverage for TIME Arrow ingestion and NULL constraints.
mssql-py-core/tests/test_bulkcopy_arrow_tinyint.py	Integration coverage for TINYINT Arrow ingestion and out-of-range handling.
mssql-py-core/tests/test_bulkcopy_arrow_transactions.py	Integration coverage for batch_size + use_internal_transaction semantics via Arrow sources.
mssql-py-core/tests/test_bulkcopy_arrow_unsupported_types.py	Integration coverage ensuring unsupported Arrow→SQL pairs fail fast with “not supported”.
mssql-py-core/tests/test_bulkcopy_arrow_uuid.py	Integration coverage for UNIQUEIDENTIFIER ingestion from 16-byte Arrow binary.
mssql-py-core/tests/test_bulkcopy_arrow_varchar.py	Integration coverage for VARCHAR ingestion incl. MAX case and NULL constraints.

[Copilot]

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated 2 comments.

[David-Engel]

Can you improve the diff coverage to >= overall coverage? We rely heavily on test automation for validation so we want to maintain a high bar there.

[saurabh500]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[David-Engel]

Better! One more push for your agent to cover more should do it. I'd suggest they make sure to include negative tests, based on the current missing lines.

[saurabh500]

@gargsaumya merge from main. Then the unit test reporting in mssql-py-core will start showing up as well in the diff coverage. That would help with the test numbers a lot