diff --git a/airbyte-cdk/java/airbyte-cdk/README.md b/airbyte-cdk/java/airbyte-cdk/README.md index a477cbce0413..d7048556de33 100644 --- a/airbyte-cdk/java/airbyte-cdk/README.md +++ b/airbyte-cdk/java/airbyte-cdk/README.md @@ -156,6 +156,7 @@ MavenLocal debugging steps: | Version | Date | Pull Request | Subject | |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------| +| 0.4.7 | 2023-11-08 | [\#31856](https://github.com/airbytehq/airbyte/pull/31856) | source-postgres: support for inifinty date and timestamps | | 0.4.5 | 2023-11-07 | [\#32112](https://github.com/airbytehq/airbyte/pull/32112) | Async destinations framework: Allow configuring the queue flush threshold | | 0.4.4 | 2023-11-06 | [\#32119](https://github.com/airbytehq/airbyte/pull/32119) | Add STANDARD UUID codec to MongoDB debezium handler | | 0.4.2 | 2023-11-06 | [\#32190](https://github.com/airbytehq/airbyte/pull/32190) | Improve error deinterpolation | diff --git a/airbyte-cdk/java/airbyte-cdk/airbyte-commons-protocol/src/test/resources/WellKnownTypes.json b/airbyte-cdk/java/airbyte-cdk/airbyte-commons-protocol/src/test/resources/WellKnownTypes.json index 95d2ff9e26fa..9e4d6656deae 100644 --- a/airbyte-cdk/java/airbyte-cdk/airbyte-commons-protocol/src/test/resources/WellKnownTypes.json +++ b/airbyte-cdk/java/airbyte-cdk/airbyte-commons-protocol/src/test/resources/WellKnownTypes.json @@ -11,18 +11,39 @@ }, "Date": { "type": "string", - "pattern": "^\\d{4}-\\d{2}-\\d{2}( BC)?$", - "description": "RFC 3339\u00a75.6's full-date format, extended with BC era support" + "oneOf": [ + { + "pattern": "^\\d{4}-\\d{2}-\\d{2}( BC)?$" + }, + { + "enum": ["Infinity", "-Infinity"] + } + ], + "description": "RFC 3339\u00a75.6's full-date format, extended with BC era support and (-)Infinity" }, "TimestampWithTimezone": { "type": "string", - "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?(Z|[+\\-]\\d{1,2}:\\d{2})( BC)?$", - "description": "An instant in time. Frequently simply referred to as just a timestamp, or timestamptz. Uses RFC 3339\u00a75.6's date-time format, requiring a \"T\" separator, and extended with BC era support. Note that we do _not_ accept Unix epochs here.\n" + "oneOf": [ + { + "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?(Z|[+\\-]\\d{1,2}:\\d{2})( BC)?$" + }, + { + "enum": ["Infinity", "-Infinity"] + } + ], + "description": "An instant in time. Frequently simply referred to as just a timestamp, or timestamptz. Uses RFC 3339\u00a75.6's date-time format, requiring a \"T\" separator, and extended with BC era support and (-)Infinity. Note that we do _not_ accept Unix epochs here.\n" }, "TimestampWithoutTimezone": { "type": "string", - "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?( BC)?$", - "description": "Also known as a localdatetime, or just datetime. Under RFC 3339\u00a75.6, this would be represented as `full-date \"T\" partial-time`, extended with BC era support.\n" + "oneOf": [ + { + "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?( BC)?$" + }, + { + "enum": ["Infinity", "-Infinity"] + } + ], + "description": "Also known as a localdatetime, or just datetime. Under RFC 3339\u00a75.6, this would be represented as `full-date \"T\" partial-time`, extended with BC era support and (-)Infinity.\n" }, "TimeWithTimezone": { "type": "string", diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/postgres/PostgresConverter.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/postgres/PostgresConverter.java index 44c91f2ecbbf..1cfdbbf8eda0 100644 --- a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/postgres/PostgresConverter.java +++ b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/postgres/PostgresConverter.java @@ -15,6 +15,7 @@ import io.airbyte.cdk.db.jdbc.DateTimeConverter; import io.airbyte.cdk.integrations.debezium.internals.DebeziumConverterUtils; +import io.debezium.connector.postgresql.PostgresValueConverter; import io.debezium.spi.converter.CustomConverter; import io.debezium.spi.converter.RelationalColumn; import io.debezium.time.Conversions; @@ -239,11 +240,13 @@ private int getTimePrecision(final RelationalColumn field) { return field.scale().orElse(-1); } + private final String POSITIVE_INFINITY_VALUE = "Infinity"; + private final String NEGATIVE_INFINITY_VALUE = "-Infinity"; + // Ref : // https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-temporal-types private void registerDate(final RelationalColumn field, final ConverterRegistration registration) { final var fieldType = field.typeName(); - registration.register(SchemaBuilder.string().optional(), x -> { if (x == null) { return DebeziumConverterUtils.convertDefaultValue(field); @@ -252,8 +255,20 @@ private void registerDate(final RelationalColumn field, final ConverterRegistrat case "TIMETZ": return DateTimeConverter.convertToTimeWithTimezone(x); case "TIMESTAMPTZ": + if (x.equals(PostgresValueConverter.NEGATIVE_INFINITY_OFFSET_DATE_TIME)) { + return NEGATIVE_INFINITY_VALUE; + } + if (x.equals(PostgresValueConverter.POSITIVE_INFINITY_OFFSET_DATE_TIME)) { + return POSITIVE_INFINITY_VALUE; + } return DateTimeConverter.convertToTimestampWithTimezone(x); case "TIMESTAMP": + if (x.equals(PostgresValueConverter.NEGATIVE_INFINITY_INSTANT)) { + return NEGATIVE_INFINITY_VALUE; + } + if (x.equals(PostgresValueConverter.POSITIVE_INFINITY_INSTANT)) { + return POSITIVE_INFINITY_VALUE; + } if (x instanceof final Long l) { if (getTimePrecision(field) <= 3) { return convertToTimestamp(Conversions.toInstantFromMillis(l)); @@ -264,6 +279,12 @@ private void registerDate(final RelationalColumn field, final ConverterRegistrat } return convertToTimestamp(x); case "DATE": + if (x.equals(PostgresValueConverter.NEGATIVE_INFINITY_LOCAL_DATE)) { + return NEGATIVE_INFINITY_VALUE; + } + if (x.equals(PostgresValueConverter.POSITIVE_INFINITY_LOCAL_DATE)) { + return POSITIVE_INFINITY_VALUE; + } if (x instanceof Integer) { return convertToDate(LocalDate.ofEpochDay((Integer) x)); } diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.java index 0e1e15797b11..6c6d8b166443 100644 --- a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.java +++ b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.java @@ -46,7 +46,9 @@ public static StateManager createStateManager(final AirbyteStateType supportedSt switch (supportedStateType) { case LEGACY: LOGGER.info("Legacy state manager selected to manage state object with type {}.", airbyteStateMessage.getType()); - return new LegacyStateManager(Jsons.object(airbyteStateMessage.getData(), DbState.class), catalog); + @SuppressWarnings("deprecation") + StateManager retVal = new LegacyStateManager(Jsons.object(airbyteStateMessage.getData(), DbState.class), catalog); + return retVal; case GLOBAL: LOGGER.info("Global state manager selected to manage state object with type {}.", airbyteStateMessage.getType()); return new GlobalStateManager(generateGlobalState(airbyteStateMessage), catalog); diff --git a/airbyte-integrations/connectors/source-postgres/build.gradle b/airbyte-integrations/connectors/source-postgres/build.gradle index 7a94eedafb03..f806e1fdabb6 100644 --- a/airbyte-integrations/connectors/source-postgres/build.gradle +++ b/airbyte-integrations/connectors/source-postgres/build.gradle @@ -13,7 +13,7 @@ java { } airbyteJavaConnector { - cdkVersionRequired = '0.4.6' + cdkVersionRequired = '0.4.7' features = ['db-sources'] useLocalCdk = false } diff --git a/airbyte-integrations/connectors/source-postgres/metadata.yaml b/airbyte-integrations/connectors/source-postgres/metadata.yaml index f1fb89ecec04..1fbe97df91ce 100644 --- a/airbyte-integrations/connectors/source-postgres/metadata.yaml +++ b/airbyte-integrations/connectors/source-postgres/metadata.yaml @@ -9,7 +9,7 @@ data: connectorSubtype: database connectorType: source definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750 - dockerImageTag: 3.2.20 + dockerImageTag: 3.2.21 dockerRepository: airbyte/source-postgres documentationUrl: https://docs.airbyte.com/integrations/sources/postgres githubIssueLabel: source-postgres diff --git a/airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java b/airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java index 476fd62ea78a..e0ea271195d7 100644 --- a/airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java +++ b/airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java @@ -30,21 +30,19 @@ import io.airbyte.protocol.models.JsonSchemaPrimitiveUtil.JsonSchemaPrimitive; import io.airbyte.protocol.models.JsonSchemaType; import java.math.BigDecimal; +import java.sql.Date; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.sql.Timestamp; -import java.time.LocalDate; -import java.time.LocalDateTime; -import java.time.LocalTime; -import java.time.OffsetDateTime; -import java.time.OffsetTime; +import java.time.*; import java.time.format.DateTimeParseException; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import org.postgresql.PGStatement; import org.postgresql.geometric.PGbox; import org.postgresql.geometric.PGcircle; import org.postgresql.geometric.PGline; @@ -67,6 +65,15 @@ public class PostgresSourceOperations extends AbstractJdbcCompatibleSourceOperat private static final Map POSTGRES_TYPE_DICT = new HashMap<>(); private final Map> streamColumnInfo = new HashMap<>(); + private static final String POSITIVE_INFINITY_STRING = "Infinity"; + private static final String NEGATIVE_INFINITY_STRING = "-Infinity"; + private static final Date POSITIVE_INFINITY_DATE = new Date(PGStatement.DATE_POSITIVE_INFINITY); + private static final Date NEGATIVE_INFINITY_DATE = new Date(PGStatement.DATE_NEGATIVE_INFINITY); + private static final Timestamp POSITIVE_INFINITY_TIMESTAMP = new Timestamp(PGStatement.DATE_POSITIVE_INFINITY); + private static final Timestamp NEGATIVE_INFINITY_TIMESTAMP = new Timestamp(PGStatement.DATE_NEGATIVE_INFINITY); + private static final OffsetDateTime POSITIVE_INFINITY_OFFSET_DATE_TIME = OffsetDateTime.MAX; + private static final OffsetDateTime NEGATIVE_INFINITY_OFFSET_DATE_TIME = OffsetDateTime.MIN; + static { Arrays.stream(PostgresType.class.getEnumConstants()).forEach(c -> POSTGRES_TYPE_DICT.put(c.type, c)); } @@ -92,6 +99,8 @@ public void setCursorField(final PreparedStatement preparedStatement, final PostgresType cursorFieldType, final String value) throws SQLException { + + LOGGER.warn("SGX setCursorField value=" + value + "cursorFieldType=" + cursorFieldType); switch (cursorFieldType) { case TIMESTAMP -> setTimestamp(preparedStatement, parameterIndex, value); @@ -265,6 +274,10 @@ private void putTimestampArray(final ObjectNode node, final String columnName, f final Timestamp timestamp = arrayResultSet.getTimestamp(2); if (timestamp == null) { arrayNode.add(NullNode.getInstance()); + } else if (POSITIVE_INFINITY_TIMESTAMP.equals(timestamp)) { + arrayNode.add(POSITIVE_INFINITY_STRING); + } else if (NEGATIVE_INFINITY_TIMESTAMP.equals(timestamp)) { + arrayNode.add(NEGATIVE_INFINITY_STRING); } else { arrayNode.add(DateTimeConverter.convertToTimestamp(timestamp)); } @@ -280,6 +293,10 @@ private void putTimestampTzArray(final ObjectNode node, final String columnName, final OffsetDateTime timestamptz = getObject(arrayResultSet, 2, OffsetDateTime.class); if (timestamptz == null) { arrayNode.add(NullNode.getInstance()); + } else if (POSITIVE_INFINITY_OFFSET_DATE_TIME.equals(timestamptz)) { + arrayNode.add(POSITIVE_INFINITY_STRING); + } else if (NEGATIVE_INFINITY_OFFSET_DATE_TIME.equals(timestamptz)) { + arrayNode.add(NEGATIVE_INFINITY_STRING); } else { final LocalDate localDate = timestamptz.toLocalDate(); arrayNode.add(resolveEra(localDate, timestamptz.format(TIMESTAMPTZ_FORMATTER))); @@ -292,9 +309,13 @@ private void putDateArray(final ObjectNode node, final String columnName, final final ArrayNode arrayNode = Jsons.arrayNode(); final ResultSet arrayResultSet = resultSet.getArray(colIndex).getResultSet(); while (arrayResultSet.next()) { - final LocalDate date = getObject(arrayResultSet, 2, LocalDate.class); + final Date date = getObject(arrayResultSet, 2, Date.class); if (date == null) { arrayNode.add(NullNode.getInstance()); + } else if (POSITIVE_INFINITY_DATE.equals(date)) { + arrayNode.add(POSITIVE_INFINITY_STRING); + } else if (NEGATIVE_INFINITY_DATE.equals(date)) { + arrayNode.add(NEGATIVE_INFINITY_STRING); } else { arrayNode.add(DateTimeConverter.convertToDate(date)); } @@ -391,9 +412,57 @@ private void putLongArray(final ObjectNode node, final String columnName, final node.set(columnName, arrayNode); } + @Override + protected void putDate(ObjectNode node, String columnName, ResultSet resultSet, int index) throws SQLException { + Date dateFromResultSet = resultSet.getDate(index); + if (POSITIVE_INFINITY_DATE.equals(dateFromResultSet)) { + node.put(columnName, POSITIVE_INFINITY_STRING); + } else if (NEGATIVE_INFINITY_DATE.equals(dateFromResultSet)) { + node.put(columnName, NEGATIVE_INFINITY_STRING); + } else { + super.putDate(node, columnName, resultSet, index); + } + } + + @Override + protected void putTime(ObjectNode node, String columnName, ResultSet resultSet, int index) throws SQLException { + super.putTime(node, columnName, resultSet, index); + } + @Override protected void putTimestamp(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException { - node.put(columnName, DateTimeConverter.convertToTimestamp(resultSet.getTimestamp(index))); + Timestamp timestampFromResultSet = resultSet.getTimestamp(index); + String strValue = resultSet.getString(index); + if (POSITIVE_INFINITY_TIMESTAMP.equals(timestampFromResultSet)) { + node.put(columnName, POSITIVE_INFINITY_STRING); + } else if (NEGATIVE_INFINITY_TIMESTAMP.equals(timestampFromResultSet)) { + node.put(columnName, NEGATIVE_INFINITY_STRING); + } else { + node.put(columnName, DateTimeConverter.convertToTimestamp(timestampFromResultSet)); + } + } + + @Override + protected void putTimeWithTimezone(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException { + final OffsetTime timetz = getObject(resultSet, index, OffsetTime.class); + node.put(columnName, DateTimeConverter.convertToTimeWithTimezone(timetz)); + } + + @Override + protected void putTimestampWithTimezone(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) + throws SQLException { + final OffsetDateTime timestampTz = getObject(resultSet, index, OffsetDateTime.class); + final String timestampTzVal; + if (POSITIVE_INFINITY_OFFSET_DATE_TIME.equals(timestampTz)) { + timestampTzVal = POSITIVE_INFINITY_STRING; + } else if (NEGATIVE_INFINITY_OFFSET_DATE_TIME.equals(timestampTz)) { + timestampTzVal = NEGATIVE_INFINITY_STRING; + } else { + final LocalDate localDate = timestampTz.toLocalDate(); + timestampTzVal = resolveEra(localDate, timestampTz.format(TIMESTAMPTZ_FORMATTER)); + } + + node.put(columnName, timestampTzVal); } @Override diff --git a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/AbstractPostgresSourceDatatypeTest.java b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/AbstractPostgresSourceDatatypeTest.java index f958c8cf3360..00c1722eef25 100644 --- a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/AbstractPostgresSourceDatatypeTest.java +++ b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/AbstractPostgresSourceDatatypeTest.java @@ -195,8 +195,8 @@ protected void initTests() { .sourceType("date") .fullSourceDataType(type) .airbyteType(JsonSchemaType.STRING_DATE) - .addInsertValues("'1999-01-08'", "'1991-02-10 BC'", "'2022/11/12'", "'1987.12.01'") - .addExpectedValues("1999-01-08", "1991-02-10 BC", "2022-11-12", "1987-12-01") + .addInsertValues("'1999-01-08'", "'1991-02-10 BC'", "'2022/11/12'", "'1987.12.01'", "'-InFinITy'", "'InFinITy'") + .addExpectedValues("1999-01-08", "1991-02-10 BC", "2022-11-12", "1987-12-01", "-Infinity", "Infinity") .build()); } @@ -460,14 +460,16 @@ protected void initTests() { "TIMESTAMP '0001-01-01 00:00:00.000000'", // The last possible timestamp in BCE "TIMESTAMP '0001-12-31 23:59:59.999999 BC'", - "'epoch'") + "'epoch'", + "'-InFinITy'", "'InFinITy'") .addExpectedValues( "2004-10-19T10:23:00.000000", "2004-10-19T10:23:54.123456", "3004-10-19T10:23:54.123456 BC", "0001-01-01T00:00:00.000000", "0001-12-31T23:59:59.999999 BC", - "1970-01-01T00:00:00.000000") + "1970-01-01T00:00:00.000000", + "-Infinity", "Infinity") .build()); } @@ -483,8 +485,6 @@ protected void initTests() { .build()); } - addTimestampWithInfinityValuesTest(); - // timestamp with time zone for (final String fullSourceType : Set.of("timestamptz", "timestamp with time zone")) { addDataTypeTestData( @@ -503,14 +503,14 @@ protected void initTests() { "TIMESTAMP WITH TIME ZONE '0001-12-31 16:00:00.000000-08 BC'", // The last possible timestamp in BCE (15:59-08 == 23:59Z) "TIMESTAMP WITH TIME ZONE '0001-12-31 15:59:59.999999-08 BC'", - "null") + "null", "'-InFinITy'", "'InFinITy'") .addExpectedValues( "2004-10-19T18:23:00.000000Z", "2004-10-19T18:23:54.123456Z", "3004-10-19T18:23:54.123456Z BC", "0001-01-01T00:00:00.000000Z", "0001-12-31T23:59:59.999999Z BC", - null) + null, "-Infinity", "Infinity") .build()); } @@ -640,24 +640,6 @@ protected void addTimeWithTimeZoneTest() { } } - protected void addTimestampWithInfinityValuesTest() { - // timestamp without time zone - for (final String fullSourceType : Set.of("timestamp", "timestamp without time zone", "timestamp without time zone not null default now()")) { - addDataTypeTestData( - TestDataHolder.builder() - .sourceType("timestamp") - .fullSourceDataType(fullSourceType) - .airbyteType(JsonSchemaType.STRING_TIMESTAMP_WITHOUT_TIMEZONE) - .addInsertValues( - "'infinity'", - "'-infinity'") - .addExpectedValues( - "+292278994-08-16T23:00:00.000000", - "+292269055-12-02T23:00:00.000000 BC") - .build()); - } - } - private void addArraysTestData() { addDataTypeTestData( TestDataHolder.builder() diff --git a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java index 8268af8ce4f1..f36fdb61b164 100644 --- a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java +++ b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java @@ -175,25 +175,6 @@ protected void addTimeWithTimeZoneTest() { } } - @Override - protected void addTimestampWithInfinityValuesTest() { - // timestamp without time zone - for (final String fullSourceType : Set.of("timestamp", "timestamp without time zone", "timestamp without time zone not null default now()")) { - addDataTypeTestData( - TestDataHolder.builder() - .sourceType("timestamp") - .fullSourceDataType(fullSourceType) - .airbyteType(JsonSchemaType.STRING_TIMESTAMP_WITHOUT_TIMEZONE) - .addInsertValues( - "'infinity'", - "'-infinity'") - .addExpectedValues( - "+294247-01-10T04:00:25.200000", - "+290309-12-21T19:59:27.600000 BC") - .build()); - } - } - @Override protected void addNumericValuesTest() { addDataTypeTestData( diff --git a/docs/integrations/sources/postgres.md b/docs/integrations/sources/postgres.md index 644e0de53b08..6c09d3aabd75 100644 --- a/docs/integrations/sources/postgres.md +++ b/docs/integrations/sources/postgres.md @@ -291,7 +291,8 @@ According to Postgres [documentation](https://www.postgresql.org/docs/14/datatyp | Version | Date | Pull Request | Subject | |---------|------------|----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| 3.2.20 | 2023-11-06 | [#32193](https://github.com/airbytehq/airbyte/pull/32193) | Adopt java CDK version 0.4.1. | +| 3.2.21 | 2023-11-07 | [31856](https://github.com/airbytehq/airbyte/pull/31856) | handle date/timestamp infinity values properly | +| 3.2.20 | 2023-11-06 | [32193](https://github.com/airbytehq/airbyte/pull/32193) | Adopt java CDK version 0.4.1. | | 3.2.19 | 2023-11-03 | [32050](https://github.com/airbytehq/airbyte/pull/32050) | Adopt java CDK version 0.4.0. | | 3.2.18 | 2023-11-01 | [29038](https://github.com/airbytehq/airbyte/pull/29038) | Fix typo (s/Airbtye/Airbyte/) | | 3.2.17 | 2023-11-01 | [32068](https://github.com/airbytehq/airbyte/pull/32068) | Bump Debezium 2.2.0Final -> 2.4.0Final |