Skip to content

NIFI-15305 Fix Date Time conversion for floating point as string #10697

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
exceptionfactory wants to merge 1 commit into
base: main
Choose a base branch
from
+73 −23
Open

NIFI-15305 Fix Date Time conversion for floating point as string #10697

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
75 changes: 55 additions & 20 deletions ...in/java/org/apache/nifi/serialization/record/field/ObjectLocalDateTimeFieldConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalAccessor;
import java.time.temporal.TemporalQueries;
import java.time.temporal.TemporalQuery;
Expand All @@ -36,6 +37,11 @@
*/
class ObjectLocalDateTimeFieldConverter implements FieldConverter<Object, LocalDateTime> {
private static final long YEAR_TEN_THOUSAND = 253_402_300_800_000L;
private static final long YEAR_TEN_THOUSAND_SECONDS = 253_402_300_800L;
private static final char PERIOD = '.';

private static final long MICROSECOND_MULTIPLIER = 1_000;
private static final long MILLISECOND_MULTIPLIER = 1_000_000;

private static final TemporalQuery<LocalDateTime> LOCAL_DATE_TIME_TEMPORAL_QUERY = new LocalDateTimeQuery();

Expand Down Expand Up @@ -66,9 +72,10 @@ public LocalDateTime convertField(final Object field, final Optional<String> pat
return ofInstant(instant);
}
case final Number number -> {
// If value is a floating point number, we consider it as seconds since epoch plus a decimal part for fractions of a second.
// Handle floating point numbers with integral and fractional components
if (field instanceof Double || field instanceof Float) {
return toLocalDateTime(number.doubleValue());
final double floatingPointNumber = number.doubleValue();
return convertDouble(floatingPointNumber);
}

return toLocalDateTime(number.longValue());
Expand Down Expand Up @@ -99,39 +106,67 @@ public LocalDateTime convertField(final Object field, final Optional<String> pat

private LocalDateTime tryParseAsNumber(final String value, final String fieldName) {
try {
// If decimal, treat as a double and convert to seconds and nanoseconds.
if (value.contains(".")) {
final double number = Double.parseDouble(value);
return toLocalDateTime(number);
final LocalDateTime localDateTime;

final int periodIndex = value.indexOf(PERIOD);
if (periodIndex >= 0) {
// Parse Double to support both decimal notation and exponent notation
final double floatingPointNumber = Double.parseDouble(value);
localDateTime = convertDouble(floatingPointNumber);
} else {
final long number = Long.parseLong(value);
localDateTime = toLocalDateTime(number);
}

// attempt to parse as a long value
final long number = Long.parseLong(value);
return toLocalDateTime(number);
return localDateTime;
} catch (final NumberFormatException e) {
throw new FieldConversionException(LocalDateTime.class, value, fieldName, e);
}
}

private LocalDateTime toLocalDateTime(final double secondsSinceEpoch) {
// Determine the number of micros past the second by subtracting the number of seconds from the decimal value and multiplying by 1 million.
final double micros = 1_000_000 * (secondsSinceEpoch - (long) secondsSinceEpoch);
// Convert micros to nanos. Note that we perform this as a separate operation, rather than multiplying by 1_000,000,000 in order to avoid
// issues that occur with rounding at high precision.
final long nanos = (long) micros * 1000L;
private LocalDateTime convertDouble(final double number) {
// Cast to long for integral part of the number
final long integral = (long) number;

// Calculate fractional part of the number for subsequent precision evaluation
final double fractional = number - integral;

return toLocalDateTime((long) secondsSinceEpoch, nanos);
return convertIntegralFractional(integral, fractional);
}

private LocalDateTime toLocalDateTime(final long epochSeconds, final long nanosPastSecond) {
final Instant instant = Instant.ofEpochSecond(epochSeconds).plusNanos(nanosPastSecond);
private LocalDateTime convertIntegralFractional(final long integral, final double fractional) {
final Instant epoch;
final long fractionalMultiplier;

if (integral > YEAR_TEN_THOUSAND_SECONDS) {
// Handle large numbers as milliseconds instead of seconds
epoch = Instant.ofEpochMilli(integral);

// Handle fractional part as number of microseconds
fractionalMultiplier = MICROSECOND_MULTIPLIER;
} else {
// Handle smaller numbers as seconds
epoch = Instant.ofEpochSecond(integral);

// Handle fractional part as number of milliseconds
fractionalMultiplier = MILLISECOND_MULTIPLIER;
}

// Calculate microseconds according to multiplier for expected precision
final double fractionalMicroseconds = fractional * fractionalMultiplier;
final long microseconds = Math.round(fractionalMicroseconds);
final Instant instant = epoch.plus(microseconds, ChronoUnit.MICROS);

return ofInstant(instant);
}

private LocalDateTime toLocalDateTime(final long value) {
if (value > YEAR_TEN_THOUSAND) {
// Value is too large. Assume microseconds instead of milliseconds.
final Instant microsInstant = Instant.ofEpochSecond(value / 1_000_000, (value % 1_000_000) * 1_000);
// Handle number as microseconds for large values
final long epochSecond = value / MILLISECOND_MULTIPLIER;
// Calculate nanoseconds from remainder
final long nanoAdjustment = (value % MILLISECOND_MULTIPLIER) * MICROSECOND_MULTIPLIER;
final Instant microsInstant = Instant.ofEpochSecond(epochSecond, nanoAdjustment);
return ofInstant(microsInstant);
}

Expand Down
21 changes: 18 additions & 3 deletions ...ava/org/apache/nifi/serialization/record/field/TestObjectLocalDateTimeFieldConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.Optional;

import static org.junit.jupiter.api.Assertions.assertEquals;
Expand All @@ -40,8 +41,15 @@ public class TestObjectLocalDateTimeFieldConverter {
private static final LocalDateTime LOCAL_DATE_TIME_MILLIS_PRECISION = LocalDateTime.ofInstant(INSTANT_MILLIS_PRECISION, ZoneId.systemDefault());
private static final LocalDateTime LOCAL_DATE_TIME_MICROS_PRECISION = LocalDateTime.ofInstant(INSTANT_MICROS_PRECISION, ZoneId.systemDefault());

private final ObjectLocalDateTimeFieldConverter converter = new ObjectLocalDateTimeFieldConverter();
private static final long INPUT_MILLISECONDS = 1765056655230L;
private static final long INPUT_MICROSECONDS = 746;
private static final int EXPECTED_NANOSECONDS = 230746000;
private static final String MILLIS_WITH_FRACTIONAL_MICROS = "%d.%d".formatted(INPUT_MILLISECONDS, INPUT_MICROSECONDS);

private static final Instant INSTANT_MICROSECOND_PRECISION = Instant.ofEpochMilli(INPUT_MILLISECONDS).plus(INPUT_MICROSECONDS, ChronoUnit.MICROS);
private static final LocalDateTime LOCAL_DATE_TIME_FRACTIONAL_MICROS = LocalDateTime.ofInstant(INSTANT_MICROSECOND_PRECISION, ZoneId.systemDefault());

private final ObjectLocalDateTimeFieldConverter converter = new ObjectLocalDateTimeFieldConverter();

@Test
public void testConvertTimestampMillis() {
Expand Down Expand Up @@ -69,8 +77,15 @@ public void testDoubleAsEpochSeconds() {
public void testDoubleAsEpochSecondsAsString() {
final LocalDateTime result = converter.convertField(MICROS_TIMESTAMP_STRING, Optional.empty(), FIELD_NAME);
assertEquals(LOCAL_DATE_TIME_MICROS_PRECISION, result);
final double expectedNanos = 351567000L;
assertEquals(expectedNanos, result.getNano(), 1D);
assertEquals(NANOS_AFTER_SECOND, result.getNano(), 1D);
}

@Test
public void testDoubleAsEpochMillisecondsWithMicroseconds() {
final LocalDateTime result = converter.convertField(MILLIS_WITH_FRACTIONAL_MICROS, Optional.empty(), FIELD_NAME);
assertEquals(LOCAL_DATE_TIME_FRACTIONAL_MICROS, result);
final int nano = result.getNano();
assertEquals(EXPECTED_NANOSECONDS, nano);
}

@Test
Expand Down
Loading