Skip to content

[split 6/22] converters: Decimal128 clamp bounds, temporal range checks, replication-situation test matrix #1361

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
minguyen9988 wants to merge 1 commit into
base: omniwatcher/split-17-ddl-parser
Choose a base branch
from
Open

[split 6/22] converters: Decimal128 clamp bounds, temporal range checks, replication-situation test matrix #1361

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions .../src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
23 changes: 14 additions & 9 deletions ...main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseDataTypeMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<MutablePair<Schema.Type, String>, ClickHouseDataType> dataTypesMap;

static {
Expand All @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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())) {
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand Down
65 changes: 49 additions & 16 deletions ...or/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}

Expand All @@ -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");
}
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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.
*
* <p>The clamp targets the <b>Decimal128</b> bounds
* (&plusmn;10<sup>38</sup>), not the wider Decimal256 bounds
* (&plusmn;10<sup>76</sup>), and that is deliberate: the clamped value
* has to survive serialization by the JDBC driver, and the driver
* checks a <em>scaled</em> 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}
* (10<sup>76</sup>, 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.</p>
*
* @param value the BigDecimal value to be truncated.
* @return the truncated BigDecimal value.
*/
Expand Down
11 changes: 2 additions & 9 deletions ...onnector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/DataTypeRange.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading
Loading