Use avro name sanitization in source (#202)

Change suggested by @ryanvanhuuksloot (Shopify)
GoogleCloudDataproc · Jan 26, 2025 · 917c2c0 · 917c2c0
1 parent d9c7983
commit 917c2c0
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 6 deletions.
diff --git a/...nnector-bigquery/src/main/java/com/google/cloud/flink/bigquery/source/BigQuerySource.java b/...nnector-bigquery/src/main/java/com/google/cloud/flink/bigquery/source/BigQuerySource.java
@@ -311,8 +311,10 @@ public static BigQuerySource<GenericRecord> readAvros(BigQueryReadOptions readOp
                                 SchemaTransform.toGenericAvroSchema(
                                                 String.format(
                                                         "%s.%s.%s",
-                                                        connectOptions.getProjectId(),
-                                                        connectOptions.getDataset(),
+                                                        sanitizeAvroSchemaName(
+                                                                connectOptions.getProjectId()),
+                                                        sanitizeAvroSchemaName(
+                                                                connectOptions.getDataset()),
                                                         sanitizeAvroSchemaName(
                                                                 connectOptions.getTable())),
                                                 tableSchema.getFields())
@@ -351,8 +353,10 @@ public static BigQuerySource<GenericRecord> streamAvros(BigQueryReadOptions read
                                 SchemaTransform.toGenericAvroSchema(
                                                 String.format(
                                                         "%s.%s.%s",
-                                                        connectOptions.getProjectId(),
-                                                        connectOptions.getDataset(),
+                                                        sanitizeAvroSchemaName(
+                                                                connectOptions.getProjectId()),
+                                                        sanitizeAvroSchemaName(
+                                                                connectOptions.getDataset()),
                                                         sanitizeAvroSchemaName(
                                                                 connectOptions.getTable())),
                                                 tableSchema.getFields())

diff --git a/pom.xml b/pom.xml
@@ -296,7 +296,7 @@ under the License.
                 <version>3.24.2</version>
                 <scope>test</scope>
             </dependency>
-            
+
             <!-- Unavoidable Flink Dependencies -->
             <!-- The Idea is to have one same flink
             version that governs annotations and flink-core version -->
@@ -306,7 +306,7 @@ under the License.
                 <artifactId>flink-annotations</artifactId>
                 <version>${flink.version}</version>
             </dependency>
-            <!-- Dependency required for SerializableSupplier, 
+            <!-- Dependency required for SerializableSupplier,
                 Preconditions, BiConsumerWithException,
                 FunctionWithException used across the Project. -->
             <dependency>