source-mssql: CDC incremental_append blocked on tables without a primary key despite valid CDC capture instance
- Langage dominant
- Python
- Étoiles
- 22.1k
- Forks
- 5.3k
- Métriques de merge des PR
- Métriques de PR en attente
Description
### Connector Name
source-mssql
### Connector Version
4.4.7
### What step the error happened?
Configuring a new connector
### Relevant information
### Summary
When CDC is enabled on a source table that has no primary key, the connector's schema discovery does not expose `_ab_cdc_*` fields and offers only `full_refresh_*` sync modes. This makes incremental_append via CDC impossible, even though it is a valid use case — CDC-based incremental append does not require a primary key, since the change table provides its own unique row ordering via `__$start_lsn` + `__$seqval`.
### Expected behaviour:
Tables with CDC enabled but no primary key should expose `_ab_cdc_*` fields and support `incremental_append` sync mode. After the initial snapshot, the connector reads exclusively from the CDC change table (`cdc.__CT`) where LSN-based ordering is sufficient for reliable incremental reads. A primary key on the source table is irrelevant to this mechanism.
### Actual behaviour:
Schema discovery silently omits `_ab_cdc_*` fields for PK-less tables. The only available sync modes are `full_refresh_append`, `full_refresh_overwrite`, and `full_refresh_overwrite_deduped`. Attempting to create a connection with `incremental_append` and CDC fields selected returns a 400 error:
▎```Invalid fields selected for stream . The list of valid field names includes: []```
Evidence that CDC is correctly configured on the source:
- `SELECT is_tracked_by_cdc FROM sys.tables WHERE name = ''` → `1`
- `cdc.change_tables` contains a capture instance for the table
- `cdc.captured_columns` lists all expected columns for the capture instance
- The only missing element is `OBJECTPROPERTY(..., 'TableHasPrimaryKey')` → `0`
### Workaround:
Adding `ALTER TABLE ... ADD RowId BIGINT IDENTITY(1,1) NOT NULL` (an identity column, not a formal PK) satisfies the connector's check. This is an unnecessary schema change on the source system.
### Request:
Decouple the PK check from `incremental_append` + CDC mode. The PK requirement is valid for `incremental_deduped_history` (which needs a key for deduplication) but should not gate `incremental_append`, where the CDC change table's LSN provides all required ordering guarantees.
### Relevant log output
```shell
2026-05-20 13:38:12,830 [public-api-executor-thread-4] ERROR i.a.c.s.e.p.AbstractThrowableProblemHandler(handle):31 - Throwable Problem Handler caught exception: │
│ io.airbyte.api.problems.throwable.generated.BadRequestProblem: Detail: class BadRequestProblemResponse { │
│ status: 400 │
│ type: https://reference.airbyte.com/reference/errors#bad-request │
│ title: bad-request │
│ detail: The request could not be understood by the server due to malformed syntax. │
│ documentationUrl: null │
│ data: class ProblemMessageData { │
│ message: Invalid fields selected for stream . The list of valid field names includes: []. │
│ } │
│ }.detail, Data: class ProblemMessageData { │
│ message: Invalid fields selected for stream . The list of valid field names includes: []. │
│ } │
│ at io.airbyte.server.apis.publicapi.helpers.AirbyteCatalogHelper.validateFieldSelection(AirbyteCatalogHelper.kt:154) │
│ at io.airbyte.server.apis.publicapi.helpers.AirbyteCatalogHelper.validateStreamConfig(AirbyteCatalogHelper.kt:318) │
│ at io.airbyte.server.apis.publicapi.helpers.AirbyteCatalogHelper.getValidConfiguredStreams(AirbyteCatalogHelper.kt:647) │
│ at io.airbyte.server.apis.publicapi.controllers.ConnectionsController.publicCreateConnection$lambda$4(ConnectionsController.kt:151) │
│ at io.airbyte.server.apis.publicapi.apiTracking.TrackingHelper.callWithTracker(TrackingHelper.kt:109) │
│ at io.airbyte.server.apis.publicapi.controllers.ConnectionsController.publicCreateConnection(ConnectionsController.kt:148) │
│ at io.airbyte.server.apis.publicapi.controllers.$ConnectionsController$Definition$Exec.dispatch(Unknown Source) │
│ at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invokeUnsafe(AbstractExecutableMethodsDefinition.java:461) │
│ at io.micronaut.context.DefaultBeanContext$BeanContextUnsafeExecutionHandle.invokeUnsafe(DefaultBeanContext.java:4454) │
│ at io.micronaut.web.router.AbstractRouteMatch.execute(AbstractRouteMatch.java:272) │
│ at io.micronaut.web.router.DefaultUriRouteMatch.execute(DefaultUriRouteMatch.java:38) │
│ at io.micronaut.http.server.RouteExecutor.executeRouteAndConvertBody(RouteExecutor.java:466) │
│ at io.micronaut.http.server.RouteExecutor.lambda$callRoute$5(RouteExecutor.java:443) │
│ at io.micronaut.core.execution.ExecutionFlow.lambda$async$0(ExecutionFlow.java:92) │
│ at io.micronaut.core.propagation.PropagatedContext.lambda$wrap$3(PropagatedContext.java:232) │
│ at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) │
│ at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) │
│ at java.base/java.lang.Thread.run(Thread.java:1583)
```
### Contribute
- [ ] Yes, I want to contribute
---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/12666
Guide de contribution
Ouvrir le guide de contribution
Évaluation
Cette issue n'a pas encore été évaluée.