source-mssql: Add SQL Server identifier quoting to fix reserved keyword issues (#69248)

Author: burakku

airbyte-integrations/connectors/source-mssql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 4.3.0-rc.4
+  dockerImageTag: 4.3.0-rc.5
   dockerRepository: airbyte/source-mssql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
   githubIssueLabel: source-mssql

airbyte-integrations/connectors/source-mssql/src/main/kotlin/io/airbyte/integrations/source/mssql/MsSqlSourceOperations.kt

@@ -297,17 +297,31 @@ class MsSqlSourceOperations :
             }
     }
 
-    fun Field.sql(): String = if (type is MsSqlServerHierarchyFieldType) "$id.ToString()" else "$id"
+    /**
+     * Quotes an identifier for SQL Server using square brackets. If the identifier contains a
+     * closing bracket ']', it will be escaped as ']]'. This protects against reserved keywords and
+     * special characters in identifier names.
+     */
+    fun String.quoted(): String = "[${this.replace("]", "]]")}]"
+
+    fun Field.sql(): String =
+        if (type is MsSqlServerHierarchyFieldType) "${id.quoted()}.ToString()" else id.quoted()
 
     fun FromNode.sql(): String =
         when (this) {
             NoFrom -> ""
-            is From -> if (this.namespace == null) "FROM $name" else "FROM $namespace.$name"
+            is From -> {
+                val ns = this.namespace
+                if (ns == null) "FROM ${name.quoted()}" else "FROM ${ns.quoted()}.${name.quoted()}"
+            }
             is FromSample -> {
+                val ns = this.namespace
                 if (sampleRateInv == 1L) {
-                    if (namespace == null) "FROM $name" else "FROM $namespace.$name"
+                    if (ns == null) "FROM ${name.quoted()}"
+                    else "FROM ${ns.quoted()}.${name.quoted()}"
                 } else {
-                    val tableName = if (namespace == null) name else "$namespace.$name"
+                    val tableName =
+                        if (ns == null) name.quoted() else "${ns.quoted()}.${name.quoted()}"
                     val samplePercent = sampleRatePercentage.toPlainString()
 
                     "FROM (SELECT TOP $sampleSize * FROM $tableName TABLESAMPLE ($samplePercent PERCENT) ORDER BY NEWID()) AS randomly_sampled"

airbyte-integrations/connectors/source-mssql/src/test/kotlin/io/airbyte/integrations/source/mssql/MsSqlServerSourceSelectQueryGeneratorTest.kt

@@ -40,7 +40,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 From("users", "dbo"),
                 limit = Limit(0),
             )
-            .assertSqlEquals("""SELECT TOP 0 id, name FROM dbo.users""")
+            .assertSqlEquals("""SELECT TOP 0 [id], [name] FROM [dbo].[users]""")
     }
 
     @Test
@@ -49,7 +49,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 SelectColumnMaxValue(Field("updated_at", OffsetDateTimeFieldType)),
                 From("orders", "dbo"),
             )
-            .assertSqlEquals("""SELECT MAX(updated_at) FROM dbo.orders""")
+            .assertSqlEquals("""SELECT MAX([updated_at]) FROM [dbo].[orders]""")
     }
 
     @Test
@@ -63,7 +63,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 ),
                 From("products", "dbo"),
             )
-            .assertSqlEquals("""SELECT id, description FROM dbo.products""")
+            .assertSqlEquals("""SELECT [id], [description] FROM [dbo].[products]""")
     }
 
     @Test
@@ -90,11 +90,11 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 Limit(1000),
             )
             .assertSqlEquals(
-                """SELECT TOP 1000 pk1, pk2, pk3, data FROM """ +
-                    """dbo.composite_table WHERE (pk1 > ?) OR """ +
-                    """((pk1 = ?) AND (pk2 > ?)) OR """ +
-                    """((pk1 = ?) AND (pk2 = ?) AND (pk3 > ?)) """ +
-                    """ORDER BY pk1, pk2, pk3""",
+                """SELECT TOP 1000 [pk1], [pk2], [pk3], [data] FROM """ +
+                    """[dbo].[composite_table] WHERE ([pk1] > ?) OR """ +
+                    """(([pk1] = ?) AND ([pk2] > ?)) OR """ +
+                    """(([pk1] = ?) AND ([pk2] = ?) AND ([pk3] > ?)) """ +
+                    """ORDER BY [pk1], [pk2], [pk3]""",
                 v1 to IntFieldType,
                 v1 to IntFieldType,
                 v2 to IntFieldType,
@@ -117,9 +117,9 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 Limit(500),
             )
             .assertSqlEquals(
-                """SELECT TOP 500 content, last_modified FROM """ +
-                    """dbo.documents """ +
-                    """WHERE (last_modified > ?) AND (last_modified <= ?) ORDER BY last_modified""",
+                """SELECT TOP 500 [content], [last_modified] FROM """ +
+                    """[dbo].[documents] """ +
+                    """WHERE ([last_modified] > ?) AND ([last_modified] <= ?) ORDER BY [last_modified]""",
                 lb to DoubleFieldType,
                 ub to DoubleFieldType,
             )
@@ -140,7 +140,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 From("employees", "hr"),
             )
             .assertSqlEquals(
-                """SELECT employee_id, org_node.ToString(), employee_name FROM hr.employees"""
+                """SELECT [employee_id], [org_node].ToString(), [employee_name] FROM [hr].[employees]"""
             )
     }
 
@@ -157,7 +157,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 From("simple_table", null),
                 limit = Limit(10),
             )
-            .assertSqlEquals("""SELECT TOP 10 col1, col2 FROM simple_table""")
+            .assertSqlEquals("""SELECT TOP 10 [col1], [col2] FROM [simple_table]""")
     }
 
     @Test
@@ -173,7 +173,7 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 Limit(10000),
             )
             .assertSqlEquals(
-                """SELECT TOP 10000 sequence_id, payload FROM dbo.events WHERE sequence_id > ? ORDER BY sequence_id""",
+                """SELECT TOP 10000 [sequence_id], [payload] FROM [dbo].[events] WHERE [sequence_id] > ? ORDER BY [sequence_id]""",
                 startValue to LongFieldType,
             )
     }
@@ -201,13 +201,30 @@ class MsSqlServerSourceSelectQueryGeneratorTest {
                 Limit(100),
             )
             .assertSqlEquals(
-                """SELECT TOP 100 id, created_at, updated_at FROM dbo.records """ +
-                    """WHERE (created_at > ?) AND (updated_at <= ?) ORDER BY created_at, updated_at""",
+                """SELECT TOP 100 [id], [created_at], [updated_at] FROM [dbo].[records] """ +
+                    """WHERE ([created_at] > ?) AND ([updated_at] <= ?) ORDER BY [created_at], [updated_at]""",
                 createdAfter to OffsetDateTimeFieldType,
                 updatedBefore to OffsetDateTimeFieldType,
             )
     }
 
+    @Test
+    fun testSelectWithReservedKeywords() {
+        // Test with reserved SQL Server keywords as column names (e.g., "End", "Start")
+        val endField = Field("End", OffsetDateTimeFieldType)
+        val startField = Field("Start", OffsetDateTimeFieldType)
+        val orderField = Field("Order", IntFieldType)
+
+        SelectQuerySpec(
+                SelectColumns(listOf(Field("Id", IntFieldType), startField, endField, orderField)),
+                From("CustomerAgreementProfiles", "dbo"),
+                limit = Limit(100),
+            )
+            .assertSqlEquals(
+                """SELECT TOP 100 [Id], [Start], [End], [Order] FROM [dbo].[CustomerAgreementProfiles]"""
+            )
+    }
+
     private fun SelectQuerySpec.assertSqlEquals(
         sql: String,
         vararg bindings: Pair<JsonNode, LosslessJdbcFieldType<*, *>>,

docs/integrations/sources/mssql.md

@@ -454,6 +454,7 @@ WHERE actor_definition_id ='b5ea17b1-f170-46dc-bc31-cc744ca984c1' AND (configura
 
 | Version    | Date       | Pull Request                                                                                                      | Subject                                                                                                                                         |
 |:-----------|:-----------|:------------------------------------------------------------------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
+| 4.3.0-rc.5 | 2025-11-08 | [69248](https://github.com/airbytehq/airbyte/pull/69248) | Add SQL Server identifier quoting to fix reserved keyword issues |
 | 4.3.0-rc.4 | 2025-11-05 | [69194](https://github.com/airbytehq/airbyte/pull/69194) | Fix composite primary key discovery to show all PK columns; separate PK discovery from sync strategy |
 | 4.3.0-rc.3 | 2025-10-31 | [69097](https://github.com/airbytehq/airbyte/pull/69097) | Fix connector state value type from string to json object                                                                                                  |
 | 4.3.0-rc.2 | 2025-10-29 | [69093](https://github.com/airbytehq/airbyte/pull/69093) | Fix state parsing error for integer and number                                                                                                  |