Source Postgres: CDC syncs failing on `interval` type columns
- 主要言語
- Python
- スター
- 22.1k
- フォーク
- 5.3k
- PR マージ指標
- PR 指標を取得中
説明
### Connector Name
source-postgres
### Connector Version
2.0.33
### What step the error happened?
During the sync
### Revelant information
We have several tables with `interval` type columns in Postgres that we are trying to sync to snowflake (destination-version 1.0.5).
These tables work when syncing postgres via the standard method, but when attempting to use replication the sync errors out. Clarification: replication-method = `Standard` in the connector, it works, but when I do replication-method = `Logical Replication (CDC)` that's when it fails.
Log attached
[airbyte_cdc_interval_error (1).txt](https://github.com/airbytehq/airbyte/files/11802461/airbyte_cdc_interval_error.1.txt)
### Relevant log output
```shell
2023-06-20 15:31:57 [44msource[0m > ERROR i.d.r.RelationalSnapshotChangeEventSource(doExecute):171 Error during snapshot java.lang.ClassCastException: class java.lang.Long cannot be cast to class org.postgresql.util.PGInterval (java.lang.Long is in module java.base of loader 'bootstrap'; org.postgresql.util.PGInterval is in unnamed module of loader 'app')
at io.airbyte.integrations.debezium.internals.postgres.PostgresConverter.lambda$registerDate$22(PostgresConverter.java:274) ~[io.airbyte.airbyte-integrations.bases-debezium-20.10.23.jar:?]
at io.debezium.relational.CustomConverterRegistry.lambda$getValueConverter$1(CustomConverterRegistry.java:147) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.relational.TableSchemaBuilder.addField(TableSchemaBuilder.java:405) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.relational.TableSchemaBuilder.lambda$create$2(TableSchemaBuilder.java:148) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[?:?]
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179) ~[?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[?:?]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[?:?]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) ~[?:?]
at io.debezium.relational.TableSchemaBuilder.create(TableSchemaBuilder.java:146) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.relational.RelationalDatabaseSchema.buildAndRegisterSchema(RelationalDatabaseSchema.java:122) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.relational.RelationalDatabaseSchema.refreshSchema(RelationalDatabaseSchema.java:192) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.connector.postgresql.PostgresSchema.lambda$refreshSchemas$2(PostgresSchema.java:146) ~[debezium-connector-postgres-2.2.0.Final.jar:2.2.0.Final]
at java.util.concurrent.ConcurrentHashMap$KeySetView.forEach(ConcurrentHashMap.java:4706) ~[?:?]
at java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1092) ~[?:?]
at io.debezium.connector.postgresql.PostgresSchema.refreshSchemas(PostgresSchema.java:146) ~[debezium-connector-postgres-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.connector.postgresql.PostgresSchema.refresh(PostgresSchema.java:90) ~[debezium-connector-postgres-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource.connectionCreated(PostgresSnapshotChangeEventSource.java:93) ~[debezium-connector-postgres-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:117) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:76) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:155) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:137) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109) ~[debezium-core-2.2.0.Final.jar:2.2.0.Final]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
at java.lang.Thread.run(Thread.java:833) ~[?:?]
Stack Trace: java.lang.ClassCastException: class java.lang.Long cannot be cast to class org.postgresql.util.PGInterval (java.lang.Long is in module java.base of loader 'bootstrap'; org.postgresql.util.PGInterval is in unnamed module of loader 'app')
at io.airbyte.integrations.debezium.internals.postgres.PostgresConverter.lambda$registerDate$22(PostgresConverter.java:274)
at io.debezium.relational.CustomConverterRegistry.lambda$getValueConverter$1(CustomConverterRegistry.java:147)
at io.debezium.relational.TableSchemaBuilder.addField(TableSchemaBuilder.java:405)
at io.debezium.relational.TableSchemaBuilder.lambda$create$2(TableSchemaBuilder.java:148)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179)
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
at io.debezium.relational.TableSchemaBuilder.create(TableSchemaBuilder.java:146)
at io.debezium.relational.RelationalDatabaseSchema.buildAndRegisterSchema(RelationalDatabaseSchema.java:122)
at io.debezium.relational.RelationalDatabaseSchema.refreshSchema(RelationalDatabaseSchema.java:192)
at io.debezium.connector.postgresql.PostgresSchema.lambda$refreshSchemas$2(PostgresSchema.java:146)
at java.base/java.util.concurrent.ConcurrentHashMap$KeySetView.forEach(ConcurrentHashMap.java:4706)
at java.base/java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1092)
at io.debezium.connector.postgresql.PostgresSchema.refreshSchemas(PostgresSchema.java:146)
at io.debezium.connector.postgresql.PostgresSchema.refresh(PostgresSchema.java:90)
at io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource.connectionCreated(PostgresSnapshotChangeEventSource.java:93)
at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:117)
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:76)
at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:155)
at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:137)
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
```
### Contribute
- [ ] Yes, I want to contribute
コントリビューションガイド
評価
この issue はまだ評価されていません。