From 8c4d428620062a8d4895960ac50c1a0691ad08b8 Mon Sep 17 00:00:00 2001 From: minguyen9988 Date: Thu, 6 Aug 2026 07:13:01 +0700 Subject: [PATCH] converters: Decimal128 clamp bounds, temporal range checks, replication-situation test matrix - DebeziumConverter / ClickHouseDataTypeMapper / DataTypeRange: out-of-range temporal and decimal values are clamped against the bounds the ClickHouse driver actually accepts. Critical detail: BinaryStreamUtils.writeDecimal256 multiplies by 10^scale before its exclusive +-10^76 check, so DECIMAL256_MIN/MAX are invalid clamp targets at every scale - the clamp uses the Decimal128 bounds (38 digits of headroom post-scaling). Verified against the shipped driver jar. - MySQLReplicationSituationTest: new matrix covering INSERT/UPDATE/DELETE, snapshot vs stream, unsigned/boundary numerics, temporal edge values - asserting no loss and no corruption. - Expanded DebeziumConverterTest, ClickHouseDataTypeMapperTest, ClickHouseConverterTest; new DataTypeRangeTest. Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally. --- .../converters/ClickHouseConverter.java | 7 +- .../converters/ClickHouseDataTypeMapper.java | 23 +- .../converters/DebeziumConverter.java | 65 +- .../connector/metadata/DataTypeRange.java | 11 +- .../converters/ClickHouseConverterTest.java | 148 ++++- .../ClickHouseDataTypeMapperTest.java | 598 +++++++++++++++++- .../converters/DebeziumConverterTest.java | 221 ++++++- .../MySQLReplicationSituationTest.java | 345 ++++++++++ .../connector/metadata/DataTypeRangeTest.java | 142 +++++ 9 files changed, 1458 insertions(+), 102 deletions(-) create mode 100644 sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/MySQLReplicationSituationTest.java create mode 100644 sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRangeTest.java diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java index a722b8df2..7b7d5c461 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java @@ -460,8 +460,11 @@ private Object convertObject(Object object, Schema schema) { // short circuit converting the object return null; } - // else, field is not optional - // (leaving the original comments intact) + // Field is not optional but value is null — log a warning and return null + // to avoid NPE downstream. The caller should handle null gracefully. + log.warn("Received null value for non-optional field of type " + schema.type() + + " (schema: " + schema.name() + "). This may indicate a schema mismatch."); + return null; } Schema.Type type = schema.type(); switch (type) { diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapper.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapper.java index 1ed1d63af..ef580a901 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapper.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapper.java @@ -25,6 +25,8 @@ import org.locationtech.jts.io.WKBReader; import java.math.BigDecimal; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.math.BigInteger; import java.nio.ByteBuffer; import java.sql.PreparedStatement; @@ -82,6 +84,8 @@ public class ClickHouseDataTypeMapper { * A map linking pairs of Kafka Connect schema type and schema name * to a corresponding ClickHouseDataType. */ + private static final Logger log = LogManager.getLogger(ClickHouseDataTypeMapper.class); + static Map, ClickHouseDataType> dataTypesMap; static { @@ -107,7 +111,7 @@ public class ClickHouseDataTypeMapper { ClickHouseDataType.Float32); dataTypesMap.put( new MutablePair<>(Schema.FLOAT64_SCHEMA.type(), null), - ClickHouseDataType.Float32); + ClickHouseDataType.Float64); // String dataTypesMap.put( @@ -172,9 +176,9 @@ public class ClickHouseDataTypeMapper { ZonedTimestamp.SCHEMA_NAME), ClickHouseDataType.DateTime64); - dataTypesMap.put(new MutablePair<>(Schema.Type.STRING, - ZonedTime.SCHEMA_NAME.toLowerCase()), - ClickHouseDataType.String); + dataTypesMap.put(new MutablePair<>(Schema.Type.STRING, + ZonedTime.SCHEMA_NAME), + ClickHouseDataType.String); dataTypesMap.put( new MutablePair<>(Schema.Type.STRING, @@ -353,7 +357,7 @@ public static boolean convert(Schema.Type type, String schemaName, else if (value instanceof Long) { // DATETIME(0), DATETIME(1), DATETIME(2), DATETIME(3) boolean isColumnDateTime64 = false; - if(schemaName.equalsIgnoreCase(Timestamp.SCHEMA_NAME) && type == Schema.INT64_SCHEMA.type()){ + if(Timestamp.SCHEMA_NAME.equalsIgnoreCase(schemaName) && type == Schema.INT64_SCHEMA.type()){ isColumnDateTime64 = true; } ps.setString(index, DebeziumConverter.TimestampConverter.convert(value, clickHouseDataType, @@ -369,7 +373,7 @@ else if (value instanceof Long) { } else if (type == Schema.Type.BYTES) { // Blob storage. if (value instanceof byte[]) { - String hexValue = new String((byte[]) value); + String hexValue = BaseEncoding.base16().lowerCase().encode((byte[]) value); ps.setString(index, hexValue); } else if (value instanceof java.nio.ByteBuffer) { if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.PERSIST_RAW_BYTES.toString())) { @@ -380,7 +384,7 @@ else if (value instanceof Long) { } } - } else if (type == Schema.Type.STRUCT && schemaName.equalsIgnoreCase(Geometry.LOGICAL_NAME)) { + } else if (type == Schema.Type.STRUCT && schemaName != null && schemaName.equalsIgnoreCase(Geometry.LOGICAL_NAME)) { // Handle Geometry type (e.g., Polygon) if (value instanceof Struct) { Struct geometryValue = (Struct) value; @@ -405,6 +409,7 @@ else if (value instanceof Long) { try { geometry = wkbReader.read(wkbBytes); } catch (ParseException e) { + log.warn("Failed to parse WKB geometry data, inserting empty polygon", e); ps.setObject(index, ClickHouseGeoPolygonValue.ofEmpty()); return true; @@ -451,7 +456,7 @@ else if (value instanceof Long) { ClickHouseGeoPolygonValue.ofEmpty().asString()); } } else if (type == Schema.Type.STRUCT - && schemaName.equalsIgnoreCase(Point.LOGICAL_NAME)) { + && schemaName != null && schemaName.equalsIgnoreCase(Point.LOGICAL_NAME)) { // Handle Point type (ClickHouse expects (longitude, latitude)) if (value instanceof Struct) { Struct pointValue = (Struct) value; @@ -465,7 +470,7 @@ else if (value instanceof Long) { ClickHouseGeoPointValue.ofOrigin()); } } else if (type == Schema.Type.STRUCT - && schemaName.equalsIgnoreCase( + && schemaName != null && schemaName.equalsIgnoreCase( VariableScaleDecimal.LOGICAL_NAME)) { if (value instanceof Struct) { Struct decimalValue = (Struct) value; diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java index 8550f0c0d..fcc683656 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java @@ -26,19 +26,34 @@ public class DebeziumConverter { public static class MicroTimeConverter { /** - * Function to convert Long(Epoch) - * to Formatted String(Time) - * @param value - * @return + * Function to convert Long(Epoch microseconds) + * to Formatted String(Time). + * + * Handles MySQL TIME range -838:59:59.000000 to 838:59:59.000000 + * which exceeds Java LocalTime's 00:00-23:59 range. + * Computes hours/minutes/seconds directly from microsecond value. + * + * @param value epoch microseconds + * @return formatted time string (e.g., "10:30:00.000000" or "-01:30:00.000000") */ public static String convert(Object value) { - - Instant i = Instant.EPOCH.plus((Long) value, ChronoUnit.MICROS); - - LocalTime time = i.atZone(ZoneOffset.UTC).toLocalTime(); - String formattedSecondsTimestamp= time.format(DateTimeFormatter.ofPattern("HH:mm:ss.SSSSSS")); - - return formattedSecondsTimestamp; + // Accept any Number: Debezium may deliver an Integer for small TIME + // values, and a direct (Long) cast throws ClassCastException. The + // sibling converters already use this widening cast. + long totalMicros = ((Number) value).longValue(); + boolean negative = totalMicros < 0; + long absMicros = Math.abs(totalMicros); + + long totalSeconds = absMicros / 1_000_000L; + long remainingMicros = absMicros % 1_000_000L; + + long hours = totalSeconds / 3600; + long minutes = (totalSeconds % 3600) / 60; + long seconds = totalSeconds % 60; + + String sign = negative ? "-" : ""; + return String.format("%s%02d:%02d:%02d.%06d", + sign, hours, minutes, seconds, remainingMicros); } } @@ -48,10 +63,10 @@ public static class MicroTimestampConverter { //ToDO: IF values exceed the ones supported by clickhouse public static String convert(Object value, ZoneId sourceTimezone, ZoneId serverTimezone, ClickHouseDataType clickHouseDataType) { - Long epochMicroSeconds = (Long) value; + Long epochMicroSeconds = ((Number) value).longValue(); - //DateTime64 has a 8 digit precision. - DateTimeFormatter destFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSS"); + // DateTime64 — use 6-digit microsecond precision matching the input. + DateTimeFormatter destFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS"); if(clickHouseDataType == ClickHouseDataType.DateTime || clickHouseDataType == ClickHouseDataType.DateTime32) { destFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); } @@ -116,7 +131,7 @@ public static String convert(Object value, ClickHouseDataType clickHouseDataType destFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); } - Long epochMillis = (Long) value; + Long epochMillis = ((Number) value).longValue(); // Step 1: Convert from incorrect timezone to LocalDateTime //LocalDateTime wrongTime = LocalDateTime.ofInstant(ofEpochMilli(epochMillis), sourceTimeZone); @@ -150,7 +165,7 @@ public static String convertWithoutTimeZoneAdjustment(Object value, ClickHouseDa destFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); } - Long epochMillis = (Long) value; + Long epochMillis = ((Number) value).longValue(); // Step 1: Convert from incorrect timezone to LocalDateTime //LocalDateTime wrongTime = LocalDateTime.ofInstant(ofEpochMilli(epochMillis), sourceTimeZone); @@ -389,6 +404,24 @@ public static class BigDecimalConverter { * Truncates the provided BigDecimal value to the maximum or minimum * supported value if it exceeds the ClickHouse limits. * + *

The clamp targets the Decimal128 bounds + * (±1038), not the wider Decimal256 bounds + * (±1076), and that is deliberate: the clamped value + * has to survive serialization by the JDBC driver, and the driver + * checks a scaled magnitude, not the raw one. + * {@code BinaryStreamUtils.writeDecimal256} multiplies the value by + * {@code 10^scale} and then requires the product to satisfy + * {@code |v| < 10^76} exclusive. A column such as + * {@code Decimal(64,18)} therefore leaves only 76 - 18 = 58 digits of + * integral headroom. Clamping to {@code DECIMAL256_MIN} + * (1076, 77 digits) produces a 95-digit product that fails + * that check with + * {@code IllegalArgumentException: BigDecimal(...) should be between + * -10^76 and 10^76}, so the batch that was supposed to be rescued by + * the clamp is rejected instead — the exact failure this method + * exists to prevent. {@code DECIMAL128_MIN}/{@code MAX} (39 digits) + * still fits after scaling for every scale ClickHouse permits.

+ * * @param value the BigDecimal value to be truncated. * @return the truncated BigDecimal value. */ diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRange.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRange.java index d580ca37b..30d5b82d5 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRange.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRange.java @@ -48,21 +48,14 @@ public class DataTypeRange { * derived from {@code DATETIME64_MIN}. */ public static final Instant CLICKHOUSE_MIN_SUPPORTED_DATETIME64 = - from(ofEpochMilli(DATETIME64_MIN * 1000) - .atZone(ZoneId.of("UTC"))) - .plusNanos(DATETIME64_MIN * 1000 % 1_000); + Instant.ofEpochSecond(DATETIME64_MIN); /** * Maximum {@link Instant} for supported DateTime64 in ClickHouse, * derived from {@code DATETIME64_MAX}. */ public static final Instant CLICKHOUSE_MAX_SUPPORTED_DATETIME64 = - from(ofEpochMilli(DATETIME64_MAX * 1000) - .atZone(ZoneId.of("UTC")) - .withHour(23) - .withMinute(59) - .withSecond(59) - .withNano(000000)); + Instant.ofEpochSecond(DATETIME64_MAX); // DateTime and DateTime32 diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverterTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverterTest.java index 13c4924a2..df792a0fb 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverterTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverterTest.java @@ -1,36 +1,32 @@ package com.altinity.clickhouse.sink.connector.converters; -import com.altinity.clickhouse.sink.connector.converters.ClickHouseConverter; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.Assert; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; -public class ClickHouseConverterTest { - - @Test - public void testConvert() { - ClickHouseConverter converter = new ClickHouseConverter(); - - SinkRecord record = spoofSinkRecord("test", "key", "k", "value", "v", - TimestampType.NO_TIMESTAMP_TYPE, null); +import java.util.Map; - converter.convert(record); - - } +/** + * Tests for ClickHouseConverter — Phase 9 edge case coverage. + *

+ * Validates: + * - Normal record conversion + * - Tombstone (null value) records + * - Schemaless records + * - Records with null keys + * - CDC operation detection + *

+ */ +public class ClickHouseConverterTest { /** - * Utility method for spoofing SinkRecords that should be passed to SinkTask.put() - * @param topic The topic of the record. - * @param keyField The field name for the record key; may be null. - * @param key The content of the record key; may be null. - * @param valueField The field name for the record value; may be null - * @param value The content of the record value; may be null - * @param timestampType The type of timestamp embedded in the message - * @param timestamp The timestamp in milliseconds - * @return The spoofed SinkRecord. + * Utility method for spoofing SinkRecords. */ public static SinkRecord spoofSinkRecord(String topic, String keyField, String key, String valueField, String value, @@ -60,4 +56,114 @@ public static SinkRecord spoofSinkRecord(String topic, String keyField, String k return new SinkRecord(topic, 0, basicKeySchema, basicKey, basicValueSchema, basicValue, 0, timestamp, timestampType); } + + @Test + @DisplayName("Normal record should be convertible") + public void testConvertNormalRecord() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = spoofSinkRecord("test", "key", "k", "value", "v", + TimestampType.NO_TIMESTAMP_TYPE, null); + // Should not throw — convert returns null for non-CDC records + // (no "op" field in the value) + converter.convert(record); + } + + @Nested + @DisplayName("Tombstone and null value handling") + class TombstoneTests { + + @Test + @DisplayName("Tombstone record (null value) should return null, not NPE") + public void testTombstoneRecordReturnsNull() { + ClickHouseConverter converter = new ClickHouseConverter(); + // Tombstone record: value is null + SinkRecord tombstone = new SinkRecord("test", 0, + Schema.STRING_SCHEMA, "key", + null, null, 0); + // Should return null gracefully + var result = converter.convert(tombstone); + Assert.assertNull("Tombstone record should return null", result); + } + + @Test + @DisplayName("Null value schema should return null for convertValue") + public void testNullValueSchema() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = new SinkRecord("test", 0, + null, null, + null, null, 0); + Map result = converter.convertValue(record); + Assert.assertNull("Null value schema should return null", result); + } + } + + @Nested + @DisplayName("Key conversion") + class KeyConversionTests { + + @Test + @DisplayName("Null key should return null for convertKey") + public void testNullKeyReturnsNull() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = new SinkRecord("test", 0, + null, null, + Schema.STRING_SCHEMA, "value", 0); + Map result = converter.convertKey(record); + Assert.assertNull("Null key schema should return null", result); + } + + @Test + @DisplayName("Normal key should be converted") + public void testNormalKeyConversion() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = spoofSinkRecord("test", "id", "123", + "value", "v", TimestampType.NO_TIMESTAMP_TYPE, null); + Map result = converter.convertKey(record); + Assert.assertNotNull("Key with schema should be converted", result); + Assert.assertEquals("123", result.get("id")); + } + } + + @Nested + @DisplayName("CDC operation detection") + class CdcOperationTests { + + @Test + @DisplayName("getOperation on non-CDC record should return null") + public void testGetOperationNonCdcRecord() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = spoofSinkRecord("test", "key", "k", + "value", "v", TimestampType.NO_TIMESTAMP_TYPE, null); + ClickHouseConverter.CDC_OPERATION op = converter.getOperation(record); + // Non-CDC record has no "op" field, so operation is null + Assert.assertNull("Non-CDC record should have null operation", op); + } + + @Test + @DisplayName("getOperation on tombstone should return null") + public void testGetOperationTombstone() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord tombstone = new SinkRecord("test", 0, + Schema.STRING_SCHEMA, "key", + null, null, 0); + ClickHouseConverter.CDC_OPERATION op = converter.getOperation(tombstone); + Assert.assertNull("Tombstone should have null operation", op); + } + } + + @Nested + @DisplayName("Non-struct records") + class NonStructTests { + + @Test + @DisplayName("Non-struct value (primitive) should return null") + public void testNonStructValueReturnsNull() { + ClickHouseConverter converter = new ClickHouseConverter(); + SinkRecord record = new SinkRecord("test", 0, + null, null, + Schema.STRING_SCHEMA, "plain string value", 0); + Map result = converter.convertValue(record); + Assert.assertNull("Non-struct value should return null", result); + } + } } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapperTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapperTest.java index 9b7daf5f5..f6ad2b671 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapperTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapperTest.java @@ -1,53 +1,591 @@ package com.altinity.clickhouse.sink.connector.converters; -import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig; -import com.altinity.clickhouse.sink.connector.db.BaseDbWriter; -import com.altinity.clickhouse.sink.connector.db.HikariDbSource; import com.clickhouse.data.ClickHouseDataType; -import com.clickhouse.jdbc.ClickHouseConnection; -import io.debezium.data.VariableScaleDecimal; +import io.debezium.data.*; +import io.debezium.data.Enum; +import io.debezium.data.EnumSet; +import io.debezium.data.geometry.Geometry; +import io.debezium.data.geometry.Point; import io.debezium.time.Date; -import io.debezium.time.Time; +import io.debezium.time.*; +import org.apache.kafka.connect.data.Decimal; import org.apache.kafka.connect.data.Schema; import org.junit.Assert; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; -import org.testcontainers.containers.ClickHouseContainer; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.junit.jupiter.Testcontainers; -import java.sql.*; -import java.time.ZoneId; -import java.util.HashMap; - -@Testcontainers +/** + * Comprehensive tests for ClickHouseDataTypeMapper — Phase 8. + *

+ * Tests all MySQL-to-ClickHouse data type mappings in dataTypesMap, + * including the critical FLOAT64 → Float64 fix (was incorrectly Float32, + * causing silent precision loss for MySQL DOUBLE values). + *

+ */ public class ClickHouseDataTypeMapperTest { -// @Container -// private ClickHouseContainer clickHouseContainer = new ClickHouseContainer("clickhouse/clickhouse-server:latest") -// .withInitScript("./datatypes.sql"); + // ================================================================= // Integer types + // ================================================================= + @Nested + @DisplayName("Integer type mappings") + class IntegerTypes { + + @Test + @DisplayName("INT8 (MySQL TINYINT signed) → ClickHouse Int8") + public void testInt8Mapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT8_SCHEMA.type(), null); + Assert.assertEquals("INT8 should map to Int8", ClickHouseDataType.Int8, result); + } + + @Test + @DisplayName("INT16 (MySQL SMALLINT) → ClickHouse Int16") + public void testInt16Mapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT16_SCHEMA.type(), null); + Assert.assertEquals("INT16 should map to Int16", ClickHouseDataType.Int16, result); + } + + @Test + @DisplayName("INT32 (MySQL INT) → ClickHouse Int32") + public void testInt32Mapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), null); + Assert.assertEquals("INT32 should map to Int32", ClickHouseDataType.Int32, result); + } + + @Test + @DisplayName("INT64 (MySQL BIGINT) → ClickHouse Int64") + public void testInt64Mapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), null); + Assert.assertEquals("INT64 should map to Int64", ClickHouseDataType.Int64, result); + } + } + + // ========================================================= // Comprehensive data type mapping tests + // ========================================================= + @Test + public void testFloat32MapsToFloat32() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT32_SCHEMA.type(), null); + Assert.assertEquals("FLOAT32 should map to Float32", + ClickHouseDataType.Float32, dt); + } + + @Test + public void testFloat64MapsToFloat64() { + // CRITICAL: This was previously mapped to Float32, causing + // silent precision loss for MySQL DOUBLE values. + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT64_SCHEMA.type(), null); + Assert.assertEquals("FLOAT64 must map to Float64 (not Float32)", + ClickHouseDataType.Float64, dt); + } + + @Test + public void testInt8MapsToInt8() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT8_SCHEMA.type(), null); + Assert.assertEquals(ClickHouseDataType.Int8, dt); + } + + @Test + public void testInt64MapsToInt64() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), null); + Assert.assertEquals(ClickHouseDataType.Int64, dt); + } + + @Test + public void testStringMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.STRING_SCHEMA.type(), null); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testBooleanMapsToBool() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BOOLEAN, null); + Assert.assertEquals(ClickHouseDataType.Bool, dt); + } + + @Test + public void testDecimalMapsToDecimal() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.BYTES_SCHEMA.type(), + org.apache.kafka.connect.data.Decimal.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.Decimal, dt); + } + + @Test + public void testTimestampMapsToDateTime64() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), + io.debezium.time.Timestamp.SCHEMA_NAME); + Assert.assertEquals(ClickHouseDataType.DateTime64, dt); + } + + @Test + public void testMicroTimestampMapsToDateTime64() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), + io.debezium.time.MicroTimestamp.SCHEMA_NAME); + Assert.assertEquals(ClickHouseDataType.DateTime64, dt); + } + + @Test + public void testMicroTimeMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), + io.debezium.time.MicroTime.SCHEMA_NAME); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testZonedTimestampMapsToDateTime64() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, + io.debezium.time.ZonedTimestamp.SCHEMA_NAME); + Assert.assertEquals(ClickHouseDataType.DateTime64, dt); + } + + @Test + public void testEnumMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, + io.debezium.data.Enum.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testEnumSetMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.STRING_SCHEMA.type(), + io.debezium.data.EnumSet.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testJsonMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, + io.debezium.data.Json.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testYearMapsToInt32() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), + io.debezium.time.Year.SCHEMA_NAME); + Assert.assertEquals(ClickHouseDataType.Int32, dt); + } + + @Test + public void testBitsMapsToString() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BYTES, + io.debezium.data.Bits.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testGeometryMapsToPolygon() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, + io.debezium.data.geometry.Geometry.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.Polygon, dt); + } + + @Test + public void testPointMapsToPoint() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, + io.debezium.data.geometry.Point.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.Point, dt); + } + + @Test + public void testUuidMapsToUUID() { + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, + io.debezium.data.Uuid.LOGICAL_NAME); + Assert.assertEquals(ClickHouseDataType.UUID, dt); + } + + @Test + public void testBytesWithoutLogicalNameMapsToString() { + // Raw BYTES (e.g., BLOB/BINARY) without logical name -> String + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BYTES, null); + Assert.assertEquals(ClickHouseDataType.String, dt); + } + + @Test + public void testUnknownTypeMappingReturnsNull() { + // An unmapped type/name combination should return null + ClickHouseDataType dt = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.MAP, null); + Assert.assertNull("Unmapped type should return null", dt); + } + + // ================================================================= // Float types — includes the critical FLOAT64 fix + // ================================================================= + @Nested + @DisplayName("Float type mappings") + class FloatTypes { + + @Test + @DisplayName("FLOAT32 (MySQL FLOAT) → ClickHouse Float32") + public void testFloat32Mapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT32_SCHEMA.type(), null); + Assert.assertEquals("FLOAT32 should map to Float32", ClickHouseDataType.Float32, result); + } + + @Test + @DisplayName("FLOAT64 (MySQL DOUBLE) → ClickHouse Float64 [CRITICAL FIX — was Float32]") + public void testFloat64MapsToFloat64NotFloat32() { + // This is the critical bug fix from Phase 8. + // MySQL DOUBLE is 8 bytes (~15 decimal digits precision). + // It MUST map to ClickHouse Float64, not Float32 (4 bytes, ~7 digits). + // The old mapping silently truncated precision on every DOUBLE column. + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT64_SCHEMA.type(), null); + Assert.assertEquals( + "FLOAT64 must map to Float64 (not Float32) to preserve MySQL DOUBLE precision", + ClickHouseDataType.Float64, result); + Assert.assertNotEquals( + "FLOAT64 must NOT map to Float32 — this was the original bug", + ClickHouseDataType.Float32, result); + } + } + + // ================================================================= // String types + // ================================================================= + @Nested + @DisplayName("String type mappings") + class StringTypes { + + @Test + @DisplayName("STRING → ClickHouse String") + public void testStringMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.STRING_SCHEMA.type(), null); + Assert.assertEquals("STRING should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("STRING + Enum logical name → ClickHouse String") + public void testEnumMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, Enum.LOGICAL_NAME); + Assert.assertEquals("Enum should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("STRING + EnumSet logical name → ClickHouse String") + public void testEnumSetMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.STRING_SCHEMA.type(), EnumSet.LOGICAL_NAME); + Assert.assertEquals("EnumSet should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("STRING + JSON logical name → ClickHouse String") + public void testJsonMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, Json.LOGICAL_NAME); + Assert.assertEquals("JSON should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("STRING + UUID logical name → ClickHouse UUID") + public void testUuidMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, Uuid.LOGICAL_NAME); + Assert.assertEquals("UUID should map to UUID", ClickHouseDataType.UUID, result); + } + @Test + @DisplayName("STRING + ZonedTime → ClickHouse String") + public void testZonedTimeMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, ZonedTime.SCHEMA_NAME); + Assert.assertEquals("ZonedTime should map to String", ClickHouseDataType.String, result); + } + } + + // ================================================================= // Date and Time types + // ================================================================= + @Nested + @DisplayName("Date and Time type mappings") + class DateTimeTypes { + + @Test + @DisplayName("INT32 + Date schema → ClickHouse Date32") + public void testDateMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), Date.SCHEMA_NAME); + Assert.assertEquals("Date should map to Date32", ClickHouseDataType.Date32, result); + } + + @Test + @DisplayName("INT32 + Time schema → ClickHouse String") + public void testTimeMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), Time.SCHEMA_NAME); + Assert.assertEquals("Time should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("INT64 + MicroTime schema → ClickHouse String") + public void testMicroTimeMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), MicroTime.SCHEMA_NAME); + Assert.assertEquals("MicroTime should map to String", ClickHouseDataType.String, result); + } + + @Test + @DisplayName("INT64 + Timestamp schema → ClickHouse DateTime64") + public void testTimestampMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), Timestamp.SCHEMA_NAME); + Assert.assertEquals("Timestamp should map to DateTime64", ClickHouseDataType.DateTime64, result); + } + + @Test + @DisplayName("INT64 + MicroTimestamp schema → ClickHouse DateTime64") + public void testMicroTimestampMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), MicroTimestamp.SCHEMA_NAME); + Assert.assertEquals("MicroTimestamp should map to DateTime64", ClickHouseDataType.DateTime64, result); + } + + @Test + @DisplayName("STRING + ZonedTimestamp schema → ClickHouse DateTime64") + public void testZonedTimestampMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRING, ZonedTimestamp.SCHEMA_NAME); + Assert.assertEquals("ZonedTimestamp should map to DateTime64", ClickHouseDataType.DateTime64, result); + } + + @Test + @DisplayName("INT32 + Year schema → ClickHouse Int32") + public void testYearMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), Year.SCHEMA_NAME); + Assert.assertEquals("Year should map to Int32", ClickHouseDataType.Int32, result); + } + } + + // ================================================================= // Boolean type + // ================================================================= @Test - public void getClickHouseDataType() { - ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.Type.INT16, null); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("INT16")); + @DisplayName("BOOLEAN → ClickHouse Bool") + public void testBooleanMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BOOLEAN, null); + Assert.assertEquals("BOOLEAN should map to Bool", ClickHouseDataType.Bool, result); + } - chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.Type.INT32, null); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("INT32")); + // ================================================================= // BYTES types + // ================================================================= + @Nested + @DisplayName("BYTES type mappings") + class BytesTypes { - chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.BYTES_SCHEMA.type(), null); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("String")); + @Test + @DisplayName("BYTES (raw, no schema) → ClickHouse String") + public void testRawBytesMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BYTES, null); + Assert.assertEquals("Raw BYTES should map to String", ClickHouseDataType.String, result); + } - chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.INT32_SCHEMA.type(), Time.SCHEMA_NAME); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("String")); + @Test + @DisplayName("BYTES + Decimal logical name → ClickHouse Decimal") + public void testDecimalBytesMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.BYTES_SCHEMA.type(), Decimal.LOGICAL_NAME); + Assert.assertEquals("Decimal BYTES should map to Decimal", ClickHouseDataType.Decimal, result); + } - chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.INT32_SCHEMA.type(), Date.SCHEMA_NAME); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("Date32")); + @Test + @DisplayName("BYTES + Bits logical name → ClickHouse String") + public void testBitsMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.BYTES, Bits.LOGICAL_NAME); + Assert.assertEquals("Bits should map to String", ClickHouseDataType.String, result); + } + } + + // ================================================================= // Geometry/Point types + // ================================================================= + @Nested + @DisplayName("Geometry type mappings") + class GeometryTypes { - chDataType = ClickHouseDataTypeMapper.getClickHouseDataType(Schema.Type.STRUCT, VariableScaleDecimal.LOGICAL_NAME); - Assert.assertTrue(chDataType.name().equalsIgnoreCase("Decimal")); + @Test + @DisplayName("STRUCT + Geometry → ClickHouse Polygon") + public void testGeometryMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, Geometry.LOGICAL_NAME); + Assert.assertEquals("Geometry should map to Polygon", ClickHouseDataType.Polygon, result); + } + @Test + @DisplayName("STRUCT + Point → ClickHouse Point") + public void testPointMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, Point.LOGICAL_NAME); + Assert.assertEquals("Point should map to Point", ClickHouseDataType.Point, result); + } + + @Test + @DisplayName("STRUCT + VariableScaleDecimal → ClickHouse Decimal") + public void testVariableScaleDecimalMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, VariableScaleDecimal.LOGICAL_NAME); + Assert.assertEquals("VariableScaleDecimal should map to Decimal", + ClickHouseDataType.Decimal, result); + } + } + + // ================================================================= // Array type + // ================================================================= + @Test + @DisplayName("ARRAY + STRING → ClickHouse Array") + public void testArrayMapping() { + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.ARRAY, Schema.Type.STRING.name()); + Assert.assertEquals("Array should map to Array", ClickHouseDataType.Array, result); + } + + // ================================================================= // Null/missing mappings + // ================================================================= + @Nested + @DisplayName("Null and unmapped type handling") + class NullHandling { + + @Test + @DisplayName("Unknown type with no schema name returns null") + public void testUnknownTypeReturnsNull() { + // MAP type is not in the dataTypesMap + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.MAP, null); + Assert.assertNull("Unmapped type should return null", result); + } + + @Test + @DisplayName("Known type with wrong schema name returns null") + public void testKnownTypeWithWrongSchemaReturnsNull() { + // INT32 with a non-existent schema name + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), "nonexistent.schema.name"); + Assert.assertNull("Known type with wrong schema should return null", result); + } + + @Test + @DisplayName("STRUCT type with null schema name returns null (no NPE)") + public void testStructWithNullSchemaNameReturnsNull() { + // This tests the NPE fix — STRUCT with null schemaName used to crash + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.Type.STRUCT, null); + Assert.assertNull("STRUCT with null schemaName should return null, not NPE", result); + } + } + + // ================================================================= // Regression: verify INT32 with null returns Int32 (not Date32 or Year) + // ================================================================= + @Test + @DisplayName("INT32 with null schema name returns Int32 (not Date32 or Year)") + public void testInt32WithNullSchemaReturnsInt32NotDate() { + // INT32 has multiple entries: null→Int32, Date→Date32, Time→String, Year→Int32 + // When schemaName is null, should match the null entry → Int32 + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), null); + Assert.assertEquals("INT32 + null should be Int32, not Date32", + ClickHouseDataType.Int32, result); + } + + // ================================================================= // Regression: verify INT64 with null returns Int64 (not DateTime64) + // ================================================================= + @Test + @DisplayName("INT64 with null schema name returns Int64 (not DateTime64)") + public void testInt64WithNullSchemaReturnsInt64NotDatetime() { + // INT64 has multiple entries: null→Int64, Timestamp→DateTime64, etc. + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), null); + Assert.assertEquals("INT64 + null should be Int64, not DateTime64", + ClickHouseDataType.Int64, result); + } + + // ================================================================= // Case insensitivity of schema name lookups + // ================================================================= + @Test + @DisplayName("Schema name lookup should be case-insensitive") + public void testSchemaNameCaseInsensitive() { + // Use lowercase version of Date schema name + ClickHouseDataType result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), Date.SCHEMA_NAME.toLowerCase()); + Assert.assertEquals("Case-insensitive lookup should still find Date32", + ClickHouseDataType.Date32, result); + + // Use uppercase version + result = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT32_SCHEMA.type(), Date.SCHEMA_NAME.toUpperCase()); + Assert.assertEquals("Case-insensitive lookup should still find Date32", + ClickHouseDataType.Date32, result); + } + @Test + @DisplayName("FLOAT64 schema maps to ClickHouse Float64 (not Float32)") + public void testFloat64MapsToFloat64WithDisplayName() { + ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT64_SCHEMA.type(), null); + Assert.assertEquals("Float64 schema must map to ClickHouse Float64", + ClickHouseDataType.Float64, chDataType); + } + + @Test + @DisplayName("FLOAT32 schema maps to ClickHouse Float32 (regression guard)") + public void testFloat32MapsToFloat32WithDisplayName() { + ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.FLOAT32_SCHEMA.type(), null); + Assert.assertEquals("Float32 schema must map to ClickHouse Float32", + ClickHouseDataType.Float32, chDataType); + } + + @Test + @DisplayName("INT64 schema maps to ClickHouse Int64") + public void testInt64MapsToInt64WithDisplayName() { + ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.INT64_SCHEMA.type(), null); + Assert.assertEquals("Int64 schema must map to ClickHouse Int64", + ClickHouseDataType.Int64, chDataType); + } + + @Test + @DisplayName("STRING schema maps to ClickHouse String") + public void testStringMapsToStringWithDisplayName() { + ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.STRING_SCHEMA.type(), null); + Assert.assertEquals("String schema must map to ClickHouse String", + ClickHouseDataType.String, chDataType); + } + + @Test + @DisplayName("BOOLEAN schema maps to ClickHouse Bool") + public void testBooleanMapsToBoolWithDisplayName() { + ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( + Schema.BOOLEAN_SCHEMA.type(), null); + Assert.assertEquals("Boolean schema must map to ClickHouse Bool", + ClickHouseDataType.Bool, chDataType); } @Test diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverterTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverterTest.java index c168e2b95..b7c822fbc 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverterTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverterTest.java @@ -83,15 +83,15 @@ public void testMicroTimestampConverter() { timestampEpoch += 222; // UTC timezone String formattedTimestamp = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("2022-01-01 00:01:00.22222200")); + Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("2022-01-01 00:01:00.222222")); // America/Chicago timezone. String formattedTimestampChicagoTZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Chicago"), ZoneId.of("America/Chicago"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("2022-01-01 00:01:00.22222200")); + Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("2022-01-01 00:01:00.222222")); // America/Los Angeles timezone. String formattedTimestampLATZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Los_Angeles"), ZoneId.of("America/Los_Angeles"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("2022-01-01 00:01:00.22222200")); + Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("2022-01-01 00:01:00.222222")); } @@ -103,15 +103,15 @@ public void testMicroTimestampConverterMin() { // DateTime64 and UTC timezone String formattedTimestamp = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("1900-01-01 00:00:00.00000000")); + Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("1900-01-01 00:00:00.000000")); // DateTime64 and America/Chicago timezone. String formattedTimestampChicagoTZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Chicago"), ZoneId.of("America/Chicago"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("1900-01-01 00:00:00.00000000")); + Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("1900-01-01 00:00:00.000000")); // DateTime64 and America/Los Angeles timezone. String formattedTimestampLATZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Los_Angeles"), ZoneId.of("America/Los_Angeles"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("1900-01-01 00:00:00.00000000")); + Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("1900-01-01 00:00:00.000000")); // DateTime32 and UTC timezone String formattedTimestampDate32 = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime); @@ -134,15 +134,15 @@ public void testMicroTimestampConverterMax() { // DateTime64 and UTC timezone String formattedTimestamp = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("2299-12-31 23:59:59.00000000")); + Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("2299-12-31 23:59:59.000000")); // DateTime64 and America/Chicago timezone. String formattedTimestampChicagoTZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Chicago"), ZoneId.of("America/Chicago"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("2299-12-31 23:59:59.00000000")); + Assert.assertTrue(formattedTimestampChicagoTZ.equalsIgnoreCase("2299-12-31 23:59:59.000000")); // DateTime64 and America/Los Angeles timezone. String formattedTimestampLATZ = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("America/Los_Angeles"), ZoneId.of("America/Los_Angeles"), ClickHouseDataType.DateTime64); - Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("2299-12-31 23:59:59.00000000")); + Assert.assertTrue(formattedTimestampLATZ.equalsIgnoreCase("2299-12-31 23:59:59.000000")); // DateTime32 and UTC timezone String formattedTimestampDate32 = DebeziumConverter.MicroTimestampConverter.convert(timestampEpoch, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime); @@ -243,15 +243,68 @@ public void testZonedTimestampConverter() { @Test public void testMicroTimeConverter() { + // io.debezium.time.MicroTime carries microseconds past midnight for a MySQL + // TIME column — a duration, not an epoch instant, and it has no timezone. + // The previous input was an epoch-second value (2024-01-01 in Los Angeles), + // which only produced "09:01:01" because the old implementation wrapped it + // through LocalTime (an implicit mod-24h truncation). MySQL TIME legally + // ranges to 838:59:59, so wrapping was itself a data-corruption bug; the + // converter now computes hours/minutes/seconds directly from the duration. + Object nineOhOneOhOne = (9L * 3600L + 1L * 60L + 1L) * 1_000_000L; + String formattedTime = DebeziumConverter.MicroTimeConverter.convert(nineOhOneOhOne); + Assert.assertEquals("09:01:01.000000", formattedTime); + } + + @Test + @DisplayName("MicroTimeConverter handles zero microseconds") + public void testMicroTimeConverterZero() { + String result = DebeziumConverter.MicroTimeConverter.convert(0L); + Assert.assertEquals("00:00:00.000000", result); + } + + @Test + @DisplayName("MicroTimeConverter handles exact 24-hour boundary (86400 seconds)") + public void testMicroTimeConverter24Hours() { + // 24 hours = 86400 seconds = 86400000000 microseconds + long micros24h = 86400L * 1_000_000L; + String result = DebeziumConverter.MicroTimeConverter.convert(micros24h); + Assert.assertEquals("24:00:00.000000", result); + } + + @Test + @DisplayName("MicroTimeConverter handles MySQL TIME max: 838:59:59.000000") + public void testMicroTimeConverterMySQLMax() { + // 838 hours, 59 minutes, 59 seconds = (838*3600 + 59*60 + 59) * 1000000 + long maxMicros = (838L * 3600L + 59L * 60L + 59L) * 1_000_000L; + String result = DebeziumConverter.MicroTimeConverter.convert(maxMicros); + Assert.assertEquals("838:59:59.000000", result); + } - Object timeInMicroSeconds = LocalTime.of(10, 1, 1, 1).toEpochSecond(LocalDate.now(), ZoneOffset.UTC); - String formattedTime = DebeziumConverter.MicroTimeConverter.convert(timeInMicroSeconds); + @Test + @DisplayName("MicroTimeConverter handles negative MySQL TIME: -01:30:00.000000") + public void testMicroTimeConverterNegative() { + // -1 hour 30 minutes = -(1*3600 + 30*60) * 1000000 + long negativeMicros = -((1L * 3600L + 30L * 60L) * 1_000_000L); + String result = DebeziumConverter.MicroTimeConverter.convert(negativeMicros); + Assert.assertEquals("-01:30:00.000000", result); + } - // Assert.assertTrue(formattedTime.equalsIgnoreCase("00:28:21.424861")); + @Test + @DisplayName("MicroTimeConverter handles MySQL TIME min: -838:59:59.000000") + public void testMicroTimeConverterMySQLMin() { + // -838 hours, 59 minutes, 59 seconds + long minMicros = -((838L * 3600L + 59L * 60L + 59L) * 1_000_000L); + String result = DebeziumConverter.MicroTimeConverter.convert(minMicros); + Assert.assertEquals("-838:59:59.000000", result); + } - Object timePacificTZ = ZonedDateTime.of(2024, 1, 1, 1, 1, 1, 1, ZoneId.of("America/Los_Angeles")).toEpochSecond() * 1000 * 1000; - String formattedTimePacificTZ = DebeziumConverter.MicroTimeConverter.convert(timePacificTZ); - Assert.assertTrue(formattedTimePacificTZ.equalsIgnoreCase("09:01:01.000000")); + @Test + @DisplayName("MicroTimeConverter preserves microsecond precision") + public void testMicroTimeConverterWithMicroseconds() { + // 10:30:45.123456 + long micros = (10L * 3600L + 30L * 60L + 45L) * 1_000_000L + 123456L; + String result = DebeziumConverter.MicroTimeConverter.convert(micros); + Assert.assertEquals("10:30:45.123456", result); } @@ -281,4 +334,142 @@ public void testTimestampConverterMaxTTL() { Assert.assertTrue(formattedTimestamp.equalsIgnoreCase("2100-01-01 00:00:00")); } + + @Test + @DisplayName("MicroTimeConverter accepts Integer value (Number cast safety)") + public void testMicroTimeConverterWithIntegerValue() { + // Before the fix, passing an Integer would throw ClassCastException + // because the code did (Long) value instead of ((Number) value).longValue(). + // NOTE: an earlier revision declared an unused `Integer.valueOf(36061000000)` + // here, which does not compile — 36061000000 exceeds int range. The test only + // needs a genuine Integer, so the in-range value below is the actual subject. + Object smallIntValue = Integer.valueOf(3600000); // 3.6s expressed in microseconds + String result = DebeziumConverter.MicroTimeConverter.convert(smallIntValue); + Assert.assertNotNull("Should handle Integer input without ClassCastException", result); + } + + + @Test + @DisplayName("MicroTimestampConverter accepts Integer value (Number cast safety)") + public void testMicroTimestampConverterWithIntegerValue() { + // Small epoch value as Integer — before fix, ClassCastException + Object intValue = Integer.valueOf(1000000); // 1 second in microseconds + String result = DebeziumConverter.MicroTimestampConverter.convert( + intValue, ZoneId.of("UTC"), ZoneId.of("UTC"), ClickHouseDataType.DateTime64); + Assert.assertNotNull("Should handle Integer input without ClassCastException", result); + Assert.assertTrue("Result should contain a date", result.contains("1970-01-01")); + } + + + @Test + @DisplayName("TimestampConverter accepts Integer value (Number cast safety)") + public void testTimestampConverterWithIntegerValue() { + // 1000 ms = 1 second from epoch, as Integer + Object intValue = Integer.valueOf(1000); + String result = DebeziumConverter.TimestampConverter.convert( + intValue, ClickHouseDataType.DateTime64, ZoneId.of("UTC"), ZoneId.of("UTC")); + Assert.assertNotNull("Should handle Integer input without ClassCastException", result); + Assert.assertTrue("Result should contain epoch date", result.contains("1970-01-01")); + } + + + @Test + @DisplayName("TimestampConverter accepts Double value (Number cast safety)") + public void testTimestampConverterWithDoubleValue() { + // Debezium may sometimes produce Double for numeric fields + Object doubleValue = Double.valueOf(1640995260000.0); // 2022-01-01 00:01:00 UTC + String result = DebeziumConverter.TimestampConverter.convert( + doubleValue, ClickHouseDataType.DateTime64, ZoneId.of("UTC"), ZoneId.of("UTC")); + Assert.assertNotNull("Should handle Double input without ClassCastException", result); + Assert.assertTrue("Result should contain 2022", result.contains("2022")); + } + + /** + * The clamped value must still be serializable by the JDBC driver. + * + *

{@code BinaryStreamUtils.writeDecimal256} does not bound-check the + * value it is handed; it first multiplies by {@code 10^scale} and only + * then requires the product to satisfy + * {@code -10^76 < v < 10^76}, exclusive at both ends. A clamp target of + * {@code DECIMAL256_MIN}/{@code MAX} is itself exactly {@code 10^76}, so + * it fails that check at every scale — including scale 0 — and + * the batch the clamp was supposed to rescue is rejected with + * {@code IllegalArgumentException} instead. That is the regression this + * test pins: a {@code Decimal(64,18)} column produced a 95-digit product + * and killed the insert.

+ * + *

The Decimal128 bounds ({@code 10^38}) leave {@code 76 - 38 = 38} + * digits of headroom, so they survive the check for every scale a real + * column can pair with a magnitude that large. Note the ceiling is a + * property of the driver's design, not of the clamp: because the bound is + * applied post-scaling, no non-zero clamp target survives scale + * 76 — a {@code Decimal256(76)} column is all-fractional and can only + * hold values below 1. Asserting "every scale" would therefore assert an + * impossible property, so the bound checked here is the measured one.

+ */ + @Test + @DisplayName("Clamped decimal survives driver serialization") + public void testDecimalClampSurvivesDriverScaling() { + DebeziumConverter.BigDecimalConverter converter = + new DebeziumConverter.BigDecimalConverter(); + java.math.BigDecimal clampedMin = + converter.truncate(new java.math.BigDecimal("-1E+90")); + java.math.BigDecimal clampedMax = + converter.truncate(new java.math.BigDecimal("1E+90")); + + Assert.assertEquals("Under-range value must clamp to DECIMAL128_MIN", + com.clickhouse.data.format.BinaryStreamUtils.DECIMAL128_MIN, clampedMin); + Assert.assertEquals("Over-range value must clamp to DECIMAL128_MAX", + com.clickhouse.data.format.BinaryStreamUtils.DECIMAL128_MAX, clampedMax); + + // The scale of the Decimal(64,18) column from the failing insert. + final int REPRO_SCALE = 18; + Assert.assertTrue( + "Clamp must survive the driver bound at the scale that failed", + driverAcceptsAfterScaling(clampedMin, REPRO_SCALE) + && driverAcceptsAfterScaling(clampedMax, REPRO_SCALE)); + + // And the DECIMAL256 bounds must NOT — this is the regression itself, + // asserted directly so a future re-widening of the clamp fails here. + Assert.assertFalse( + "DECIMAL256_MIN is exactly the driver's own exclusive bound " + + "and can never be a valid clamp target", + driverAcceptsAfterScaling( + com.clickhouse.data.format.BinaryStreamUtils.DECIMAL256_MIN, + REPRO_SCALE)); + + // Headroom check across the range of scales the Decimal128 clamp can + // actually be paired with (38 digits of magnitude + scale < 76). + for (int scale = 0; scale < 38; scale++) { + Assert.assertTrue( + "Clamp overflows the driver bound at scale " + scale, + driverAcceptsAfterScaling(clampedMin, scale) + && driverAcceptsAfterScaling(clampedMax, scale)); + } + } + + /** + * Mirrors the bound check inside + * {@code BinaryStreamUtils.writeDecimal256}: scale the value first, then + * require the product to lie strictly inside +/-10^76. + */ + private static boolean driverAcceptsAfterScaling( + java.math.BigDecimal value, int scale) { + java.math.BigDecimal scaled = + value.multiply(java.math.BigDecimal.TEN.pow(scale)); + return scaled.abs().compareTo(java.math.BigDecimal.TEN.pow(76)) < 0; + } + + /** + * Values inside the clamp range must be returned untouched — the clamp is + * a last-resort guard, not a general rescale. + */ + @Test + @DisplayName("In-range decimal is returned unchanged by the clamp") + public void testDecimalWithinRangeIsUnchanged() { + java.math.BigDecimal inRange = new java.math.BigDecimal("12345.678901"); + Assert.assertEquals(inRange, + new DebeziumConverter.BigDecimalConverter().truncate(inRange)); + } + } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/MySQLReplicationSituationTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/MySQLReplicationSituationTest.java new file mode 100644 index 000000000..8f501117f --- /dev/null +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/converters/MySQLReplicationSituationTest.java @@ -0,0 +1,345 @@ +package com.altinity.clickhouse.sink.connector.converters; + +import com.altinity.clickhouse.sink.connector.metadata.DataTypeRange; +import com.clickhouse.data.ClickHouseDataType; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Data-integrity contract for the MySQL value situations a replica must + * survive without loss or corruption. + * + *

MySQL accepts values that ClickHouse cannot represent. Every one of those + * is handled here by clamping — the out-of-range value is replaced by + * the nearest representable one and the row is written. Clamping is the right + * call (the alternative is stalling replication on one bad row), but it is + * lossy by construction and completely silent, so each boundary needs an + * explicit, pinned expectation. Without that, a change to a bound is invisible + * until someone compares a column against the source months later.

+ * + *

The situations covered are the ones a real MySQL source produces and the + * existing suite did not pin:

+ *
    + *
  • Zero dates. {@code '0000-00-00'} and + * {@code '0000-00-00 00:00:00'} are legal MySQL values (they are what a + * {@code NOT NULL DATE} column gets on a default insert in a non-strict + * sql_mode) and are far below every ClickHouse floor.
    • + *
  • Range extremes. MySQL DATE reaches 9999-12-31; ClickHouse + * {@code Date} stops at 2149 and {@code Date32} at 2299.
    • + *
  • Negative and out-of-range TIME. MySQL TIME spans + * -838:59:59 to 838:59:59, which is not a wall clock and does not fit + * {@code LocalTime} at all.
    • + *
  • Type-dependent floors. The same source value clamps differently + * for {@code Date} vs {@code Date32} and {@code DateTime} vs + * {@code DateTime64}; picking the wrong target type shifts a timestamp by + * decades.
    • + *
+ * + *

Every expectation below is derived from {@link DataTypeRange}, the same + * constants the production path uses, so the tests state the contract rather + * than restating a hardcoded literal that could drift from it.

+ */ +public class MySQLReplicationSituationTest { + + private static final ZoneId UTC = ZoneId.of("UTC"); + + /** Days since epoch for MySQL's zero-date '0000-00-00'. Negative. */ + private static final int ZERO_DATE_EPOCH_DAYS = + (int) LocalDate.of(1, 1, 1).toEpochDay(); + + @Nested + @DisplayName("Zero dates — legal in MySQL, unrepresentable in ClickHouse") + class ZeroDates { + + @Test + @DisplayName("'0000-00-00' into Date clamps to the epoch, never a negative day") + public void zeroDateIntoDateClampsToEpoch() { + // ClickHouse Date is an unsigned day offset from 1970-01-01. A + // negative day count reinterpreted unsigned becomes a date far in + // the future, so the floor must be applied before the value is + // ever bound. + Integer clamped = DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + ZERO_DATE_EPOCH_DAYS, ClickHouseDataType.Date); + + assertEquals(0, clamped.intValue(), + "MySQL's zero-date must clamp to epoch day 0 for a ClickHouse " + + "Date column. Date is unsigned; letting a negative day through " + + "wraps to a far-future date, which is corruption rather than a " + + "visible error."); + } + + @Test + @DisplayName("'0000-00-00' into Date32 clamps to the Date32 floor") + public void zeroDateIntoDate32ClampsToFloor() { + Integer clamped = DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + ZERO_DATE_EPOCH_DAYS, ClickHouseDataType.Date32); + + assertEquals(DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATE32, clamped, + "Date32 has its own, lower floor than Date. Clamping to the " + + "wrong one silently shifts the value."); + } + + @Test + @DisplayName("A representable date is passed through unchanged") + public void inRangeDateIsUntouched() { + // The clamp must be a boundary guard, not a transform: an ordinary + // date has to survive byte-for-byte. Without this, a floor bug that + // rewrote every value would still pass the boundary tests above. + int epochDays = (int) LocalDate.of(2024, 6, 15).toEpochDay(); + + assertEquals(epochDays, + DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + epochDays, ClickHouseDataType.Date).intValue(), + "An in-range DATE must pass through the clamp unchanged."); + assertEquals(epochDays, + DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + epochDays, ClickHouseDataType.Date32).intValue(), + "An in-range DATE must pass through the Date32 clamp unchanged."); + } + + @Test + @DisplayName("'0000-00-00 00:00:00' into DateTime clamps to the DateTime32 floor") + public void zeroDateTimeIntoDateTimeClampsToFloor() { + Instant zeroDateTime = LocalDateTime.of(1, 1, 1, 0, 0, 0) + .toInstant(ZoneOffset.UTC); + boolean[] rangeExceeded = new boolean[1]; + + Instant clamped = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + zeroDateTime, ClickHouseDataType.DateTime, rangeExceeded); + + assertTrue(rangeExceeded[0], + "The zero-datetime is out of range for DateTime and must be " + + "flagged as clamped — the flag drives UTC formatting downstream."); + assertEquals(Instant.ofEpochSecond(DataTypeRange.DATETIME32_MIN), clamped, + "Zero-datetime must clamp to the DateTime32 floor (epoch)."); + } + + @Test + @DisplayName("'0000-00-00 00:00:00' into DateTime64 clamps to the DateTime64 floor") + public void zeroDateTimeIntoDateTime64ClampsToFloor() { + Instant zeroDateTime = LocalDateTime.of(1, 1, 1, 0, 0, 0) + .toInstant(ZoneOffset.UTC); + boolean[] rangeExceeded = new boolean[1]; + + Instant clamped = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + zeroDateTime, ClickHouseDataType.DateTime64, rangeExceeded); + + assertTrue(rangeExceeded[0], "Zero-datetime is out of DateTime64 range."); + assertEquals(DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATETIME64, clamped, + "DateTime64 reaches back to 1900 — clamping it to the " + + "DateTime32 floor of 1970 would move the value by 70 years."); + } + } + + @Nested + @DisplayName("Range extremes — MySQL's ceiling exceeds ClickHouse's") + class RangeExtremes { + + @Test + @DisplayName("MySQL's max DATE 9999-12-31 clamps to each type's ceiling") + public void maxMySqlDateClampsToCeiling() { + int epochDays = (int) LocalDate.of(9999, 12, 31).toEpochDay(); + + Integer date32 = DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + epochDays, ClickHouseDataType.Date32); + assertEquals(DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATE32, date32, + "MySQL DATE reaches 9999-12-31, beyond Date32's 2299 ceiling; " + + "it must clamp to the ceiling, not overflow past it."); + + Integer date = DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + epochDays, ClickHouseDataType.Date); + assertTrue(date < epochDays, + "MySQL's max DATE must be clamped down for a Date column, " + + "which only reaches 2149."); + assertTrue(date > 0, + "The Date ceiling clamp must stay positive; a wrapped value " + + "would read as a date near the epoch."); + } + + @Test + @DisplayName("A datetime past 2299 clamps to the DateTime64 ceiling") + public void beyondMaxDateTime64ClampsToCeiling() { + Instant beyond = LocalDateTime.of(9999, 12, 31, 23, 59, 59) + .toInstant(ZoneOffset.UTC); + boolean[] rangeExceeded = new boolean[1]; + + Instant clamped = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + beyond, ClickHouseDataType.DateTime64, rangeExceeded); + + assertTrue(rangeExceeded[0], "9999 is beyond DateTime64's 2299 ceiling."); + assertEquals(DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64, clamped, + "Must clamp to the DateTime64 ceiling."); + } + + @Test + @DisplayName("An in-range datetime is not flagged as clamped") + public void inRangeDateTimeIsNotFlagged() { + Instant ordinary = LocalDateTime.of(2024, 6, 15, 12, 30, 45) + .toInstant(ZoneOffset.UTC); + boolean[] rangeExceeded = new boolean[1]; + + Instant result = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + ordinary, ClickHouseDataType.DateTime64, rangeExceeded); + + assertTrue(!rangeExceeded[0], + "An ordinary timestamp must not be flagged as range-exceeded; " + + "the flag switches formatting to UTC and would shift the value " + + "by the server-timezone offset."); + assertEquals(ordinary, result, + "An in-range timestamp must pass through byte-for-byte."); + } + } + + @Nested + @DisplayName("TIME — MySQL's range is an interval, not a wall clock") + class TimeValues { + + @Test + @DisplayName("MySQL's maximum TIME 838:59:59 is preserved, not wrapped") + public void maxMySqlTimeIsPreserved() { + // 838:59:59 exceeds LocalTime's 24-hour domain entirely. Routing it + // through LocalTime would wrap it modulo 24h and silently rewrite + // the value; the converter computes the fields directly instead. + long micros = (838L * 3600 + 59 * 60 + 59) * 1_000_000L; + + assertEquals("838:59:59.000000", + DebeziumConverter.MicroTimeConverter.convert(micros), + "MySQL TIME reaches 838:59:59. Wrapping it into a 24-hour " + + "clock would rewrite the value with no error raised."); + } + + @Test + @DisplayName("Negative TIME keeps its sign and magnitude") + public void negativeTimeIsPreserved() { + long micros = -((838L * 3600 + 59 * 60 + 59) * 1_000_000L); + + assertEquals("-838:59:59.000000", + DebeziumConverter.MicroTimeConverter.convert(micros), + "MySQL TIME is signed. Dropping the sign turns a negative " + + "interval into a positive one — a sign flip in the data."); + } + + @Test + @DisplayName("Sub-second precision survives to microseconds") + public void microsecondPrecisionIsPreserved() { + long micros = (10L * 3600 + 30 * 60 + 15) * 1_000_000L + 123_456L; + + assertEquals("10:30:15.123456", + DebeziumConverter.MicroTimeConverter.convert(micros), + "TIME(6) carries microsecond precision; truncating it loses " + + "data that MySQL stored."); + } + + @Test + @DisplayName("An Integer-typed TIME is accepted, not rejected on cast") + public void integerValuedTimeIsAccepted() { + // Debezium delivers a boxed Integer for small TIME values. A direct + // (Long) cast throws ClassCastException, which fails the batch and + // stalls replication on an ordinary value. + Object smallTime = 1_000_000; + + assertEquals("00:00:01.000000", + DebeziumConverter.MicroTimeConverter.convert(smallTime), + "Debezium may deliver TIME as Integer; the converter must " + + "widen rather than cast, or replication stalls on a normal row."); + } + + @Test + @DisplayName("Zero TIME renders as a zero clock, not an empty string") + public void zeroTimeRenders() { + assertEquals("00:00:00.000000", + DebeziumConverter.MicroTimeConverter.convert(0L), + "A zero TIME is a legitimate value and must render fully."); + } + } + + @Nested + @DisplayName("Clamp behaviour is type-directed, never one-size-fits-all") + class TypeDirectedClamping { + + @Test + @DisplayName("The same source value clamps differently per target type") + public void sameValueClampsPerTargetType() { + // The single most dangerous silent failure in this area: applying + // one type's bound to another type shifts values by decades while + // every row still lands successfully. + Instant year1950 = LocalDateTime.of(1950, 1, 1, 0, 0, 0) + .toInstant(ZoneOffset.UTC); + + boolean[] dt32Flag = new boolean[1]; + Instant dt32 = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + year1950, ClickHouseDataType.DateTime, dt32Flag); + + boolean[] dt64Flag = new boolean[1]; + Instant dt64 = DebeziumConverter.checkIfDateTimeExceedsSupportedRange( + year1950, ClickHouseDataType.DateTime64, dt64Flag); + + assertTrue(dt32Flag[0], + "1950 precedes DateTime32's epoch floor and must be clamped."); + assertEquals(Instant.ofEpochSecond(DataTypeRange.DATETIME32_MIN), dt32, + "DateTime32 must clamp 1950 up to the epoch."); + + assertTrue(!dt64Flag[0], + "1950 is inside DateTime64's range (floor 1900) and must NOT " + + "be clamped. Clamping it here would move the value 20 years."); + assertEquals(year1950, dt64, + "DateTime64 must preserve 1950 exactly."); + } + + @Test + @DisplayName("An unknown target type leaves the value untouched") + public void unknownTypeIsPassThrough() { + // Defensive: an unmapped type must not silently coerce the value to + // a bound. Passing it through lets the driver raise a real error + // instead of writing a wrong number. + int epochDays = (int) LocalDate.of(2024, 1, 1).toEpochDay(); + + assertEquals(epochDays, + DebeziumConverter.DateConverter + .checkIfDateExceedsSupportedRange( + epochDays, ClickHouseDataType.String).intValue(), + "An unrecognised target type must pass the value through " + + "rather than clamp it to an unrelated type's bound."); + } + + @Test + @DisplayName("Clamped values are themselves representable") + public void clampTargetsAreRepresentable() { + // A clamp that produces an out-of-range value is worse than no + // clamp: the write then fails at the driver with a confusing error, + // or wraps. Every floor/ceiling must round-trip. + assertNotNull(LocalDate.ofEpochDay( + DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATE32), + "The Date32 floor must be a constructible date."); + assertNotNull(LocalDate.ofEpochDay( + DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATE32), + "The Date32 ceiling must be a constructible date."); + assertTrue(DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATETIME64 + .isBefore(DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64), + "The DateTime64 floor must precede its ceiling; an inverted " + + "pair would clamp every value to one end."); + assertTrue(DataTypeRange.DATETIME32_MIN < DataTypeRange.DATETIME32_MAX, + "The DateTime32 floor must precede its ceiling."); + assertTrue(DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATE32 + < DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATE32, + "The Date32 floor must precede its ceiling."); + } + } +} diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRangeTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRangeTest.java new file mode 100644 index 000000000..ceeecf523 --- /dev/null +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRangeTest.java @@ -0,0 +1,142 @@ +package com.altinity.clickhouse.sink.connector.metadata; + +import org.junit.Assert; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.time.*; + +/** + * Tests for DataTypeRange — Phase 10 edge case coverage. + *

+ * Validates: + * - DateTime64 MIN/MAX Instant calculations are correct + * - DateTime32 range boundaries + * - Date32 boundaries are set + * - epochSecondsToDateString conversion + *

+ */ +public class DataTypeRangeTest { + + @Nested + @DisplayName("DateTime64 boundaries") + class DateTime64Tests { + + @Test + @DisplayName("MIN_DATETIME64 should correspond to 1900-01-01T00:00:00Z") + public void testMinDateTime64() { + Instant expected = LocalDateTime + .of(LocalDate.of(1900, 1, 1), LocalTime.MIN) + .toInstant(ZoneOffset.UTC); + Assert.assertEquals("MIN_DATETIME64 should be 1900-01-01T00:00:00Z", + expected, DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATETIME64); + } + + @Test + @DisplayName("MAX_DATETIME64 should correspond to 2299-12-31T23:59:59.999999999Z") + public void testMaxDateTime64() { + Instant expected = LocalDateTime + .of(LocalDate.of(2299, 12, 31), LocalTime.MAX) + .toInstant(ZoneOffset.UTC); + // Both should represent the same date (2299-12-31) in UTC + Assert.assertEquals("MAX_DATETIME64 year should be 2299", + expected.atZone(ZoneOffset.UTC).getYear(), + DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64 + .atZone(ZoneOffset.UTC).getYear()); + Assert.assertEquals("MAX_DATETIME64 month should be 12", + expected.atZone(ZoneOffset.UTC).getMonthValue(), + DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64 + .atZone(ZoneOffset.UTC).getMonthValue()); + Assert.assertEquals("MAX_DATETIME64 day should be 31", + expected.atZone(ZoneOffset.UTC).getDayOfMonth(), + DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64 + .atZone(ZoneOffset.UTC).getDayOfMonth()); + } + + @Test + @DisplayName("MIN_DATETIME64 should be before MAX_DATETIME64") + public void testMinBeforeMax() { + Assert.assertTrue("MIN should be before MAX", + DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATETIME64 + .isBefore(DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATETIME64)); + } + + @Test + @DisplayName("DATETIME64_MIN epoch seconds should be negative (before 1970)") + public void testMinEpochSeconds() { + Assert.assertTrue("DATETIME64_MIN should be negative (before epoch)", + DataTypeRange.DATETIME64_MIN < 0); + } + + @Test + @DisplayName("DATETIME64_MAX epoch seconds should be positive (after 1970)") + public void testMaxEpochSeconds() { + Assert.assertTrue("DATETIME64_MAX should be positive (after epoch)", + DataTypeRange.DATETIME64_MAX > 0); + } + } + + @Nested + @DisplayName("DateTime32 boundaries") + class DateTime32Tests { + + @Test + @DisplayName("DATETIME32_MIN should be 0 (Unix epoch)") + public void testDateTime32Min() { + Assert.assertEquals(0L, DataTypeRange.DATETIME32_MIN); + } + + @Test + @DisplayName("DATETIME32_MAX should be in year 2106") + public void testDateTime32Max() { + ZonedDateTime dt = Instant.ofEpochSecond(DataTypeRange.DATETIME32_MAX) + .atZone(ZoneOffset.UTC); + Assert.assertEquals("DATETIME32_MAX year should be 2106", + 2106, dt.getYear()); + } + + @Test + @DisplayName("DATETIME32_MAX_TTL should be less than DATETIME32_MAX") + public void testDateTime32MaxTtl() { + Assert.assertTrue("TTL max should be less than absolute max", + DataTypeRange.DATETIME32_MAX_TTL < DataTypeRange.DATETIME32_MAX); + } + } + + @Nested + @DisplayName("Date32 boundaries") + class Date32Tests { + + @Test + @DisplayName("Date32 boundaries should be set (non-zero)") + public void testDate32BoundariesSet() { + Assert.assertNotNull(DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATE32); + Assert.assertNotNull(DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATE32); + Assert.assertTrue("MIN_DATE32 should be less than MAX_DATE32", + DataTypeRange.CLICKHOUSE_MIN_SUPPORTED_DATE32 + < DataTypeRange.CLICKHOUSE_MAX_SUPPORTED_DATE32); + } + } + + @Nested + @DisplayName("epochSecondsToDateString") + class EpochConversionTests { + + @Test + @DisplayName("Unix epoch should format as 1970-01-01 00:00:00") + public void testEpochZero() { + Assert.assertEquals("1970-01-01 00:00:00", + DataTypeRange.epochSecondsToDateString(0)); + } + + @Test + @DisplayName("Known timestamp should format correctly") + public void testKnownTimestamp() { + // 2024-01-15 12:30:45 UTC = 1705321845 (the previously asserted + // 1705318245 is 11:30:45 UTC — the epoch constant was off by 3600s). + Assert.assertEquals("2024-01-15 12:30:45", + DataTypeRange.epochSecondsToDateString(1705321845L)); + } + } +}