This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push: new f338215851 NIFI-13529 Set Calcite Connection timeZone to UTC for Records f338215851 is described below commit f338215851bcaec40b673af4ef4600b478a5c9f3 Author: Mark Payne <marka...@hotmail.com> AuthorDate: Tue Jul 9 11:01:41 2024 -0400 NIFI-13529 Set Calcite Connection timeZone to UTC for Records Calcite adjusts Timestamp objects returned from a Result Set based on the configured Time Zone Offset in an attempt to localize the results. Framework use of Calcite for Record processing expects input Timestamp values to remain unchanged, so setting the timeZone property to UTC with an offset of 0 effectively avoids this Calcite localization. This closes #9066 Signed-off-by: David Handermann <exceptionfact...@apache.org> --- .../java/org/apache/nifi/sql/CalciteDatabase.java | 4 +++ .../org/apache/nifi/sql/TestCalciteDatabase.java | 40 ++++++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/nifi-commons/nifi-calcite-utils/src/main/java/org/apache/nifi/sql/CalciteDatabase.java b/nifi-commons/nifi-calcite-utils/src/main/java/org/apache/nifi/sql/CalciteDatabase.java index f5d9574400..537f9956f3 100644 --- a/nifi-commons/nifi-calcite-utils/src/main/java/org/apache/nifi/sql/CalciteDatabase.java +++ b/nifi-commons/nifi-calcite-utils/src/main/java/org/apache/nifi/sql/CalciteDatabase.java @@ -193,6 +193,10 @@ public class CalciteDatabase implements Closeable { calciteProperties = properties; } + // If not explicitly set, default timezone to UTC. We ensure that when we provide timestamps, we convert them to UTC. We don't want + // Calcite trying to convert them again. + calciteProperties.putIfAbsent("timeZone", "UTC"); + final Connection sqlConnection = DriverManager.getConnection("jdbc:calcite:", calciteProperties); final CalciteConnection connection = sqlConnection.unwrap(CalciteConnection.class); connection.getRootSchema().setCacheEnabled(false); diff --git a/nifi-commons/nifi-calcite-utils/src/test/java/org/apache/nifi/sql/TestCalciteDatabase.java b/nifi-commons/nifi-calcite-utils/src/test/java/org/apache/nifi/sql/TestCalciteDatabase.java index 9d06bd6e84..b8604bc28a 100644 --- a/nifi-commons/nifi-calcite-utils/src/test/java/org/apache/nifi/sql/TestCalciteDatabase.java +++ b/nifi-commons/nifi-calcite-utils/src/test/java/org/apache/nifi/sql/TestCalciteDatabase.java @@ -24,6 +24,8 @@ import java.io.IOException; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.sql.Timestamp; +import java.util.ArrayList; import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -85,6 +87,25 @@ public class TestCalciteDatabase { } } + @Test + public void testWithTimestamp() throws SQLException, IOException { + final String query = "SELECT * FROM CANNED_DATA"; + + try (final CalciteDatabase database = createNameTimestampDatabase(); + final PreparedStatement stmt = database.getConnection().prepareStatement(query); + final ResultSet resultSet = stmt.executeQuery()) { + + assertTrue(resultSet.next()); + + // We should get the same result whether we call getTimestamp() or getObject(). We should also get back the same original Long value. + final Timestamp timestamp = resultSet.getTimestamp(2); + assertEquals(timestamp, resultSet.getObject(2)); + assertEquals(1704056400000L, timestamp.getTime()); + + assertFalse(resultSet.next()); + } + } + public static class ToUpperCase { public String invoke(final String value) { return value.toUpperCase(); @@ -113,6 +134,25 @@ public class TestCalciteDatabase { return database; } + private CalciteDatabase createNameTimestampDatabase() throws SQLException { + final CalciteDatabase database = new CalciteDatabase(); + + final NiFiTableSchema tableSchema = new NiFiTableSchema(List.of( + new ColumnSchema("name", String.class, false), + new ColumnSchema("dob", Timestamp.class, false) + )); + + final List<Object[]> rows = new ArrayList<>(); + rows.add(new Object[] {"Mark", new Timestamp(1704056400000L)}); + + final ListDataSource arrayListDataSource = new ListDataSource(tableSchema, rows); + + final NiFiTable table = new NiFiTable("CANNED_DATA", arrayListDataSource, mock(ComponentLog.class)); + database.addTable(table); + + return database; + } + private static class ListDataSource implements ResettableDataSource { private final NiFiTableSchema schema;