airbytehq / airbytehq/airbyte

Source Postgres: CDC syncs failing on `interval` type columns

オープン
#27,510 コメント 3 件 リアクション 1 件 担当者 0 名 GitHub で見る
area/connectors connectors/source/postgres frozen team/db-dw-sources type/bug
主要言語
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 source > 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 はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。