[destination-s3-data-lake]: No lowercase column names, breaks Snowflake; potentially other engines using Glue catalog
- Dominant language
- Python
- Stars
- 22.1k
- Forks
- 5.3k
- PR merge metrics
- PR metrics pending
Description
### Connector Name
destination-s3-data-lake
### Connector Version
0.3.44
### What step the error happened?
Other
### Relevant information
The `destination-s3-data-lake` connector preserves source column casing verbatim (e.g. `userId`, `createdAt`) when writing Iceberg table schemas. There is no option to control this behaviour.
A common pattern with this connector is to land data in S3/Glue and query it from downstream engines. Some of these engines require lowercase identifiers when reading Iceberg tables through a Glue catalog. Snowflake is one such engine:
From the [Snowflake documentation](https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table-glue):
> For databases linked to AWS Glue, you must use lowercase letters and surround the schema, table, and column names in double quotes. This is also required for other Iceberg REST catalogs that only support lowercase identifiers.
Tables written by this connector with mixed-case column names are unreadable from these engines, with no workaround available.
## Why external workarounds are not viable
Renaming columns after the fact (e.g. via PyIceberg schema evolution) does not work for ongoing syncs. `IcebergTableSynchronizer.maybeApplySchemaChanges()` runs at the start of every sync and compares the incoming schema (mixed-case, derived from the source) against the current table schema. Any externally lowercased columns are treated as removed, and the original mixed-case columns are re-added. For CDC workloads this makes post-hoc fixes completely unworkable.
## Root cause
The connector does not provide a `TableSchemaMapper` implementation. The CDK falls back to `NoopTableSchemaMapper`, which passes all names through unchanged:
```kotlin
// NoopTableSchemaMapper.kt
@Singleton
@Secondary
class NoopTableSchemaMapper : TableSchemaMapper {
override fun toColumnName(name: String) = name
// ...
}
```
`IcebergUtil.toIcebergSchema()` then builds the Iceberg schema directly from the raw source column names:
```kotlin
// IcebergUtil.kt
fun toIcebergSchema(stream: DestinationStream): Schema {
val schema = ObjectType(LinkedHashMap(stream.tableSchema.columnSchema.inputSchema))
return schema.withAirbyteMeta(true).toIcebergSchema(primaryKeys)
}
```
And the connector uses this schema as-is:
```kotlin
// S3DataLakeStreamLoader.kt
private val incomingSchema = icebergUtil.toIcebergSchema(stream = stream)
```
There is no configuration option, environment variable, or runtime flag to control column name casing.
## Existing infrastructure in the CDK
The CDK already has full support for column name transformation. `JsonConverter` maps record field names through `TableSchemaMapper.toColumnName()` at parse time:
```kotlin
// JsonConverter.kt
val mappedKey = enriched.stream.tableSchema.getFinalColumnName(field.key)
```
The `ColumnNameResolver`, `TableSchemaMapper` interface, and `inputToFinalColumnNames` mapping are all in place. A connector just needs to provide a `TableSchemaMapper` implementation and apply the mapped names to the Iceberg schema.
## Suggestion
Add a configurable option (e.g. `lowercase_column_names: true`) to the connector specification that, when enabled, lowercases all column names written to the Iceberg schema.
## Implementation precedent in destination-gcs-data-lake
The CDK infrastructure to support this already exists. The GCS Data Lake connector uses it to solve a different but structurally identical problem: BigLake rejects column names containing special characters. It provides a `GcsDataLakeTableSchemaMapper` that sanitises names via `Transformations.toAlphanumericAndUnderscore()`, and a `transformSchemaWithMappedNames()` method in its stream loader that applies `inputToFinalColumnNames` to the Iceberg schema after construction.
The S3 Data Lake connector has neither of these.
### Relevant log output
```shell
```
### Contribute
- [x] Yes, I want to contribute
---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/11856
Contributor guide
Assessment
This issue has not been assessed yet.