From 06b04994576bdd15824fab83e52b6692a5f48191 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Fri, 16 Oct 2026 16:14:24 +0200 Subject: [PATCH 1/4] Auto-detect the JDBC server timezone for datetime incremental offsets, falling back to the default timezone if detection fails. --- .../co/absa/pramen/api/sql/SqlGenerator.scala | 8 ++- .../pramen/api/sql/SqlGeneratorBase.scala | 60 ++++++++++++++++--- .../core/bookkeeper/OffsetManagerUtils.scala | 15 +++++ .../pramen/core/reader/TableReaderJdbc.scala | 9 ++- .../core/reader/TableReaderJdbcBase.scala | 23 ++++++- .../core/reader/TableReaderJdbcNative.scala | 10 +++- .../reader/model/TableReaderJdbcConfig.scala | 8 ++- .../core/sql/SqlGeneratorDatabricks.scala | 9 ++- .../pramen/core/sql/SqlGeneratorDb2.scala | 9 ++- .../pramen/core/sql/SqlGeneratorDenodo.scala | 9 ++- .../pramen/core/sql/SqlGeneratorGeneric.scala | 9 ++- .../pramen/core/sql/SqlGeneratorHive.scala | 9 ++- .../pramen/core/sql/SqlGeneratorHsqlDb.scala | 9 ++- .../core/sql/SqlGeneratorMicrosoft.scala | 60 ++++++++++++++++--- .../pramen/core/sql/SqlGeneratorMySQL.scala | 9 ++- .../pramen/core/sql/SqlGeneratorOracle.scala | 9 ++- .../core/sql/SqlGeneratorPostgreSQL.scala | 9 ++- .../pramen/core/sql/SqlGeneratorSas.scala | 9 ++- .../pramen/core/mocks/SqlGeneratorDummy.scala | 8 ++- .../sql/SqlGeneratorDatabricksSuite.scala | 7 ++- .../core/tests/sql/SqlGeneratorDb2Suite.scala | 7 ++- .../tests/sql/SqlGeneratorDenodoSuite.scala | 7 ++- .../tests/sql/SqlGeneratorGenericSuite.scala | 35 ++++++----- .../tests/sql/SqlGeneratorHiveSuite.scala | 7 ++- .../tests/sql/SqlGeneratorHsqlSuite.scala | 7 ++- .../sql/SqlGeneratorMicrosoftSuite.scala | 47 ++++++++------- .../tests/sql/SqlGeneratorMySqlSuite.scala | 7 ++- .../tests/sql/SqlGeneratorOracleSuite.scala | 7 ++- .../sql/SqlGeneratorPostgreSqlSuite.scala | 7 ++- .../core/tests/sql/SqlGeneratorSasSuite.scala | 7 ++- 30 files changed, 311 insertions(+), 125 deletions(-) diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala index 10dd17176..b1b854173 100644 --- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala +++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala @@ -19,7 +19,8 @@ package za.co.absa.pramen.api.sql import za.co.absa.pramen.api.offset.OffsetValue import java.sql.Connection -import java.time.LocalDate +import java.time.{LocalDate, ZoneId} +import scala.util.Try trait SqlGenerator { /** @@ -76,7 +77,10 @@ trait SqlGenerator { onlyForInfoDate: Option[LocalDate], offsetFrom: Option[OffsetValue], offsetTo: Option[OffsetValue], - columns: Seq[String]): String + columns: Seq[String], + serverTimeZone: Option[ZoneId]): String + + def getServerTimezone(connection: Connection): Try[ZoneId] /** * Returns WHERE condition for table that has the information date field given the time period. diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala index 04b8a3b24..0bb2a9d4c 100644 --- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala +++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala @@ -18,9 +18,11 @@ package za.co.absa.pramen.api.sql import za.co.absa.pramen.api.offset.{OffsetInfo, OffsetValue} -import java.time.LocalDate +import java.sql.Connection import java.time.format.DateTimeFormatter +import java.time.{LocalDate, ZoneId} import scala.collection.mutable.ListBuffer +import scala.util.Try /** * This class contains implementation of methods that are common across all SQL dialects. @@ -40,6 +42,11 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { */ def beginEndEscapeChars: (Char, Char) + /** + * Returns a query that returns the connection timezone + */ + def getServerTimezoneQuery: String + def quoteSingleIdentifier(identifier: String): String = { val (escapeBegin, escapeEnd) = beginEndEscapeChars @@ -60,7 +67,7 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { } } - def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String + def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String override def getSchemaQuery(tableName: String, columns: Seq[String]): String = { val dataQuery = getDataQuery(tableName, columns, None) @@ -68,6 +75,42 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { s"$dataQuery WHERE 0=1" } + /** + * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` + * against the given connection and parsing the first column of the first row as a timezone identifier. + * + * The statement and result set used to run the query are always closed, regardless of the outcome. + * + * @param connection An open JDBC connection to the database server whose timezone should be determined. + * @return A `Success` containing the server's `ZoneId` if the query returns a valid timezone identifier, + * or a `Failure` if the query fails, returns no rows, returns NULL, or returns a value that cannot + * be parsed as a `ZoneId`. + */ + override def getServerTimezone(connection: Connection): Try[ZoneId] = { + Try { + val query = getServerTimezoneQuery + val statement = connection.createStatement() + try { + val resultSet = statement.executeQuery(query) + try { + if (resultSet.next()) { + val timezoneStr = resultSet.getString(1) + if (timezoneStr == null) { + throw new IllegalStateException(s"The server timezone query returned NULL: $query") + } + ZoneId.of(timezoneStr.trim) + } else { + throw new IllegalStateException(s"The server timezone query returned no rows: $query") + } + } finally { + resultSet.close() + } + } finally { + statement.close() + } + } + } + override def getAliasExpression(expression: String, alias: String): String = { s"$expression AS ${escape(alias)}" } @@ -94,7 +137,8 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { onlyForInfoDate: Option[LocalDate], offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue], - columns: Seq[String]): String = { + columns: Seq[String], + serverTimeZone: Option[ZoneId]): String = { if (sqlConfig.offsetInfo.isEmpty) throw new IllegalArgumentException(s"Offset information is not configured for database table: $tableName.") @@ -103,7 +147,7 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { case None => getDataQuery(tableName, columns, None) } - val offsetWhere = getOffsetWhereClause(sqlConfig.offsetInfo.get, offsetFromOpt, offsetToOpt) + val offsetWhere = getOffsetWhereClause(sqlConfig.offsetInfo.get, offsetFromOpt, offsetToOpt, serverTimeZone) if (offsetWhere.nonEmpty) { if (onlyForInfoDate.isEmpty) { @@ -116,20 +160,20 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { } } - def getOffsetWhereClause(offsetInfo: OffsetInfo, offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue]): String = { + def getOffsetWhereClause(offsetInfo: OffsetInfo, offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue], serverTimeZone: Option[ZoneId]): String = { val offsetColumn = escape(offsetInfo.offsetColumn) (offsetFromOpt, offsetToOpt) match { case (Some(offsetFrom), Some(offsetTo)) => validateOffsetValue(offsetFrom) validateOffsetValue(offsetTo) - s"${getOffsetWhereCondition(offsetColumn, ">=", offsetFrom)} AND ${getOffsetWhereCondition(offsetColumn, "<=", offsetTo)}" + s"${getOffsetWhereCondition(offsetColumn, ">=", offsetFrom, serverTimeZone)} AND ${getOffsetWhereCondition(offsetColumn, "<=", offsetTo, serverTimeZone)}" case (Some(offsetFrom), None) => validateOffsetValue(offsetFrom) - s"${getOffsetWhereCondition(offsetColumn, ">", offsetFrom)}" + s"${getOffsetWhereCondition(offsetColumn, ">", offsetFrom, serverTimeZone)}" case (None, Some(offsetTo)) => validateOffsetValue(offsetTo) - s"${getOffsetWhereCondition(offsetColumn, "<=", offsetTo)}" + s"${getOffsetWhereCondition(offsetColumn, "<=", offsetTo, serverTimeZone)}" case (None, None) => "" } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/OffsetManagerUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/OffsetManagerUtils.scala index 8f4709a69..67f76cb79 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/OffsetManagerUtils.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/OffsetManagerUtils.scala @@ -56,7 +56,22 @@ object OffsetManagerUtils { /** + * Computes the minimum and maximum offset values present in the given DataFrame for the specified offset column. * + * For Kafka offsets, the column is expected to be a struct containing `partition` and `offset` fields. + * The minimum and maximum offsets are calculated per partition. If a current maximum offset is provided, + * the resulting values are merged with it, so partitions missing from the data take their offsets from + * the current maximum offset. + * + * For other offset types (datetime, integral, string), the minimum and maximum values of the column are + * computed, parsed into offset values and validated. + * + * @param df The DataFrame containing the ingested data. + * @param offsetColumn The name of the column that holds offset values. + * @param offsetType The type of the offset, which determines how values are aggregated and parsed. + * @param currentMaxOffset The current maximum offset, if any. For Kafka offsets it is used to fill in partitions not present in the data. + * @return `None` if the DataFrame is empty, otherwise `Some` tuple of the minimum and maximum offset values. + * @throws IllegalArgumentException if a non-Kafka offset value cannot be parsed or fails validation. */ def getMinMaxValueFromData(df: DataFrame, offsetColumn: String, offsetType: OffsetType, currentMaxOffset: Option[OffsetValue]): Option[(OffsetValue, OffsetValue)] = { diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala index 356c664e3..c71c2fa90 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala @@ -20,7 +20,7 @@ import com.typesafe.config.Config import org.apache.spark.sql.{DataFrame, SparkSession} import org.slf4j.LoggerFactory import za.co.absa.pramen.api.Query -import za.co.absa.pramen.api.offset.OffsetValue +import za.co.absa.pramen.api.offset.{OffsetType, OffsetValue} import za.co.absa.pramen.core.config.Keys import za.co.absa.pramen.core.reader.model.TableReaderJdbcConfig import za.co.absa.pramen.core.utils._ @@ -123,7 +123,12 @@ class TableReaderJdbc(jdbcReaderConfig: TableReaderJdbcConfig, offsetFrom: Option[OffsetValue], offsetTo: Option[OffsetValue], columns: Seq[String]): DataFrame = { - val sql = sqlGen.getDataQueryIncremental(tableName, onlyForInfoDate, offsetFrom, offsetTo, columns) + val tzOpt = if (offsetFrom.exists(_.dataType == OffsetType.DateTimeType) || offsetTo.exists(_.dataType == OffsetType.DateTimeType)) { + Option(serverTimeZone) + } else { + None + } + val sql = sqlGen.getDataQueryIncremental(tableName, onlyForInfoDate, offsetFrom, offsetTo, columns, tzOpt) log.info(s"JDBC Query: $sql") diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala index 8dafcd603..40ccb7caa 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala @@ -26,6 +26,7 @@ import za.co.absa.pramen.core.utils.JdbcNativeUtils.JDBC_WORDS_TO_REDACT import za.co.absa.pramen.core.utils.{ConfigUtils, JdbcNativeUtils} import java.time.LocalDate +import scala.util.{Failure, Success} abstract class TableReaderJdbcBase(jdbcReaderConfig: TableReaderJdbcConfig, jdbcUrlSelector: JdbcUrlSelector, @@ -43,18 +44,36 @@ abstract class TableReaderJdbcBase(jdbcReaderConfig: TableReaderJdbcConfig, val gen = SqlGeneratorLoader.getSqlGenerator(jdbcReaderConfig.jdbcConfig.driver, getSqlConfig) if (gen.requiresConnection) { - val (connection, _) = jdbcUrlSelector.getNewConnection(jdbcRetries) + val (connection, _) = jdbcUrlSelector.getConnection gen.setConnection(connection) } gen } + private[core] lazy val serverTimeZone = { + jdbcReaderConfig.serverTimeZone match { + case Some(tz) => + log.info(s"The server timezone is specified explicitly: $tz") + tz + case None => + val (connection, _) = jdbcUrlSelector.getConnection + sqlGen.getServerTimezone(connection) match { + case Success(tz) => + log.info(s"The server timezone is detected as: $tz") + tz + case Failure(ex) => + log.warn(s"Failed to detect the server timezone: '${ex.getMessage}'. Using the default timezone: ${jdbcReaderConfig.defaultTimeZone}") + jdbcReaderConfig.defaultTimeZone + } + } + } + private[core] def getSqlConfig: SqlConfig = { SqlConfig(jdbcReaderConfig.infoDateColumn, jdbcReaderConfig.infoDateType, jdbcReaderConfig.infoDateFormat, jdbcReaderConfig.offsetInfoOpt, - jdbcReaderConfig.serverTimeZone, + jdbcReaderConfig.defaultTimeZone, jdbcReaderConfig.identifierQuotingPolicy, jdbcReaderConfig.sqlGeneratorClass, ConfigUtils.getExtraConfig(conf, "sql")) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala index 04cfd338f..8dc741450 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala @@ -20,7 +20,7 @@ import com.typesafe.config.Config import org.apache.spark.sql.{DataFrame, SparkSession} import org.slf4j.LoggerFactory import za.co.absa.pramen.api.Query -import za.co.absa.pramen.api.offset.OffsetValue +import za.co.absa.pramen.api.offset.{OffsetType, OffsetValue} import za.co.absa.pramen.core.expr.DateExprEvaluator import za.co.absa.pramen.core.reader.model.{JdbcConfig, TableReaderJdbcConfig} import za.co.absa.pramen.core.utils._ @@ -130,7 +130,13 @@ class TableReaderJdbcNative(jdbcReaderConfig: TableReaderJdbcConfig, offsetFrom: Option[OffsetValue], offsetTo: Option[OffsetValue], columns: Seq[String]): DataFrame = { - val sql = sqlGen.getDataQueryIncremental(tableName, onlyForInfoDate, offsetFrom, offsetTo, columns) + val tzOpt = if (offsetFrom.exists(_.dataType == OffsetType.DateTimeType) || offsetTo.exists(_.dataType == OffsetType.DateTimeType)) { + Option(serverTimeZone) + } else { + None + } + + val sql = sqlGen.getDataQueryIncremental(tableName, onlyForInfoDate, offsetFrom, offsetTo, columns, tzOpt) log.info(s"JDBC Query: $sql") var df = JdbcNativeUtils.getJdbcNativeDataFrame(jdbcUrlSelector, sql) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala index a83d29f94..a4ea11e14 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala @@ -40,7 +40,8 @@ case class TableReaderJdbcConfig( enableSchemaMetadata: Boolean = false, useJdbcNative: Boolean = false, specialCharacters: String = " ", - serverTimeZone: ZoneId = ZoneId.systemDefault(), + serverTimeZone: Option[ZoneId] = None, + defaultTimeZone: ZoneId = ZoneId.systemDefault(), identifierQuotingPolicy: QuotingPolicy = QuotingPolicy.Auto, sqlGeneratorClass: Option[String] = None ) @@ -93,8 +94,8 @@ object TableReaderJdbcConfig { val offsetInfoOpt = OffsetInfoParser.fromConfig(conf) - val defaultTimezone = ConfigUtils.getOptionString(workflowConf, Keys.TIMEZONE).getOrElse("Africa/Johannesburg") - val serverTimezone = ZoneId.of(ConfigUtils.getOptionString(conf, SERVER_TIMEZONE).getOrElse(defaultTimezone)) + val defaultTimezone = ZoneId.of(ConfigUtils.getOptionString(workflowConf, Keys.TIMEZONE).getOrElse("Africa/Johannesburg")) + val serverTimezone = ConfigUtils.getOptionString(conf, SERVER_TIMEZONE).map(z =>ZoneId.of(z)) val identifierQuotingPolicy = ConfigUtils.getOptionString(conf, IDENTIFIER_QUOTING_POLICY) .map(s => QuotingPolicy.fromString(s)) @@ -131,6 +132,7 @@ object TableReaderJdbcConfig { useJdbcNative = useJdbcNative, specialCharacters, serverTimezone, + defaultTimezone, identifierQuotingPolicy = identifierQuotingPolicy, sqlGeneratorClass = ConfigUtils.getOptionString(conf, SQL_GENERATOR_CLASS_KEY) ) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala index 0db4857e7..51e2a6b5a 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala @@ -23,7 +23,7 @@ import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.DatabricksDialect import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} object SqlGeneratorDatabricks { private val log = LoggerFactory.getLogger(this.getClass) @@ -71,6 +71,8 @@ class SqlGeneratorDatabricks(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SET spark.sql.session.timeZone" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -108,10 +110,11 @@ class SqlGeneratorDatabricks(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala index 7f23a7830..b3a271eaa 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT CURRENT TIMEZONE FROM SYSIBM.SYSDUMMY1" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -94,10 +96,11 @@ class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition TIMESTAMP('$tsLiteral')" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala index 2e66342d0..f7788d0e1 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala @@ -23,7 +23,7 @@ import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.DenodoDialect import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalTime, ZonedDateTime} +import java.time.{LocalDate, LocalTime, ZoneId, ZonedDateTime} object SqlGeneratorDenodo { private val log = LoggerFactory.getLogger(this.getClass) @@ -76,6 +76,8 @@ class SqlGeneratorDenodo(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT CURRENT_TIMESTAMP AS ts, GETTIMEINMILLIS(CURRENT_TIMESTAMP) / 1000 AS epoch_sec" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { if (sqlConfig.infoDateType == SqlColumnType.DATE && dateBegin == dateEnd) { val dateBeginLit = getDateLiteral(dateBegin) @@ -107,10 +109,11 @@ class SqlGeneratorDenodo(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val zdt = ZonedDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val zdt = ZonedDateTime.ofInstant(ts, tz) val tsLiteral = timestampDbFormatter.format(zdt) s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala index 9749b1afe..6007f2afa 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorGeneric(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorGeneric(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConf s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT current_setting('TIMEZONE') AS timezone" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -94,10 +96,11 @@ class SqlGeneratorGeneric(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConf } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala index 8dad2ccb4..b27c31c24 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala @@ -23,7 +23,7 @@ import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.HiveDialect import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} object SqlGeneratorHive { private val log = LoggerFactory.getLogger(this.getClass) @@ -71,6 +71,8 @@ class SqlGeneratorHive(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT current_timestamp AS ts, unix_timestamp() AS epoch_sec" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -108,10 +110,11 @@ class SqlGeneratorHive(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala index 38b63ae21..6b360a645 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorHsqlDb(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorHsqlDb(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "VALUES CURRENT_TIMESTAMP" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -94,10 +96,11 @@ class SqlGeneratorHsqlDb(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala index 1f44b9e88..9aa23bc59 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala @@ -21,9 +21,11 @@ import za.co.absa.pramen.api.sql.SqlGeneratorBase.{needsEscaping, validateIdenti import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGenerator} import za.co.absa.pramen.core.utils.MutableStack +import java.sql.Connection import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} import scala.collection.mutable.ListBuffer +import scala.util.Try class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -72,6 +74,44 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { s"SELECT ${getLimit(limit)}${columnExpr(columns)} FROM ${escape(tableName)} WITH (NOLOCK) WHERE $where" } + def getServerTimezoneQuery: String = "SELECT CURRENT_TIMEZONE()" + + /** + * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` + * against the given connection and parsing the first column of the first row as a timezone identifier. + * + * The statement and result set used to run the query are always closed, regardless of the outcome. + * + * @param connection An open JDBC connection to the database server whose timezone should be determined. + * @return A `Success` containing the server's `ZoneId` if the query returns a valid timezone identifier, + * or a `Failure` if the query fails, returns no rows, returns NULL, or returns a value that cannot + * be parsed as a `ZoneId`. + */ + override def getServerTimezone(connection: Connection): Try[ZoneId] = { + Try { + val query = getServerTimezoneQuery + val statement = connection.createStatement() + try { + val resultSet = statement.executeQuery(query) + try { + if (resultSet.next()) { + val timezoneStr = resultSet.getString(1) + if (timezoneStr == null) { + throw new IllegalStateException(s"The server timezone query returned NULL: $query") + } + ZoneId.of(timezoneStr.trim) + } else { + throw new IllegalStateException(s"The server timezone query returned no rows: $query") + } + } finally { + resultSet.close() + } + } finally { + statement.close() + } + } + } + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -139,7 +179,8 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { onlyForInfoDate: Option[LocalDate], offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue], - columns: Seq[String]): String = { + columns: Seq[String], + serverTimeZone: Option[ZoneId]): String = { if (sqlConfig.offsetInfo.isEmpty) throw new IllegalArgumentException(s"Offset information is not configured for database table: $tableName.") @@ -148,7 +189,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { case None => getDataQuery(tableName, columns, None) } - val offsetWhere = getOffsetWhereClause(sqlConfig.offsetInfo.get, offsetFromOpt, offsetToOpt) + val offsetWhere = getOffsetWhereClause(sqlConfig.offsetInfo.get, offsetFromOpt, offsetToOpt, serverTimeZone) if (offsetWhere.nonEmpty) { if (onlyForInfoDate.isEmpty) { @@ -161,29 +202,30 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { } } - private[core] def getOffsetWhereClause(offsetInfo: OffsetInfo, offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue]): String = { + private[core] def getOffsetWhereClause(offsetInfo: OffsetInfo, offsetFromOpt: Option[OffsetValue], offsetToOpt: Option[OffsetValue], serverTimeZone: Option[ZoneId]): String = { val offsetColumn = escape(offsetInfo.offsetColumn) (offsetFromOpt, offsetToOpt) match { case (Some(offsetFrom), Some(offsetTo)) => validateOffsetValue(offsetFrom) validateOffsetValue(offsetTo) - s"${getOffsetWhereCondition(offsetColumn, ">=", offsetFrom)} AND ${getOffsetWhereCondition(offsetColumn, "<=", offsetTo)}" + s"${getOffsetWhereCondition(offsetColumn, ">=", offsetFrom, serverTimeZone)} AND ${getOffsetWhereCondition(offsetColumn, "<=", offsetTo, serverTimeZone)}" case (Some(offsetFrom), None) => validateOffsetValue(offsetFrom) - s"${getOffsetWhereCondition(offsetColumn, ">", offsetFrom)}" + s"${getOffsetWhereCondition(offsetColumn, ">", offsetFrom, serverTimeZone)}" case (None, Some(offsetTo)) => validateOffsetValue(offsetTo) - s"${getOffsetWhereCondition(offsetColumn, "<=", offsetTo)}" + s"${getOffsetWhereCondition(offsetColumn, "<=", offsetTo, serverTimeZone)}" case (None, None) => "" } } - private[core] def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + private[core] def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampMsDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala index 834602610..3181a142e 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT @@session.time_zone" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { if (sqlConfig.infoDateType == SqlColumnType.DATETIME) { s"$infoDateColumn >= '$dateBegin 00:00:00' AND $infoDateColumn < '${dateEnd.plusDays(1)} 00:00:00'" @@ -90,10 +92,11 @@ class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala index 5338761ba..a13e18835 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorOracle(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorOracle(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit, hasWhere = true)}" } + override def getServerTimezoneQuery: String = "SELECT SESSIONTIMEZONE FROM dual" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { require(!dateEnd.isBefore(dateBegin), s"infoDateEnd [$dateEnd] must be >= infoDateBegin [$dateBegin]") @@ -87,10 +89,11 @@ class SqlGeneratorOracle(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala index 9b10fc888..0a5dab1b5 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} class SqlGeneratorPostgreSQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -57,6 +57,8 @@ class SqlGeneratorPostgreSQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } + override def getServerTimezoneQuery: String = "SELECT current_setting('TIMEZONE') AS timezone" + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { if (sqlConfig.infoDateType == SqlColumnType.DATETIME) { s"$infoDateColumn >= '$dateBegin' AND $infoDateColumn < '${dateEnd.plusDays(1)}'" @@ -90,10 +92,11 @@ class SqlGeneratorPostgreSQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala index 6708f9227..bed31e3cd 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala @@ -24,7 +24,7 @@ import za.co.absa.pramen.core.sql.dialects.SasDialect import java.sql.{Connection, ResultSet} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId} import java.util.Locale import scala.collection.mutable.ListBuffer @@ -85,6 +85,8 @@ class SqlGeneratorSas(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) s"SELECT ${getColumnSql(tableName, columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit, hasWhere = true)}" } + override def getServerTimezoneQuery: String = "proc options option=timezone; run;" + private def getColumnSql(tableName: String, columns: Seq[String]): String = { if (columns.isEmpty) { getColumns(tableName).map(name => s"${escape(name)} '${escape(name)}'").mkString(", ") @@ -189,10 +191,11 @@ class SqlGeneratorSas(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) limit.map(n => s" LIMIT $n").getOrElse("") } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue): String = { + override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val ldt = LocalDateTime.ofInstant(ts, sqlConfig.serverTimeZone) + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampSasDbFormatter.format(ldt) s"$column $condition '$tsLiteral'dt" case OffsetValue.IntegralValue(value) => diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala index e26dce9d1..61d22fa66 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala @@ -19,7 +19,9 @@ package za.co.absa.pramen.core.mocks import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlConfig, SqlGenerator} -import java.time.LocalDate +import java.sql.Connection +import java.time.{LocalDate, ZoneId} +import scala.util.Try class SqlGeneratorDummy(sqlConfig: SqlConfig) extends SqlGenerator { def getSqlConfig: SqlConfig = sqlConfig @@ -38,7 +40,9 @@ class SqlGeneratorDummy(sqlConfig: SqlConfig) extends SqlGenerator { override def getDataQuery(tableName: String, infoDateBegin: LocalDate, infoDateEnd: LocalDate, columns: Seq[String], limit: Option[Int]): String = null - override def getDataQueryIncremental(tableName: String, onlyForInfoDate: Option[LocalDate], offsetFrom: Option[OffsetValue], offsetTo: Option[OffsetValue], columns: Seq[String]): String = null + override def getServerTimezone(connection: Connection): Try[ZoneId] = null + + override def getDataQueryIncremental(tableName: String, onlyForInfoDate: Option[LocalDate], offsetFrom: Option[OffsetValue], offsetTo: Option[OffsetValue], columns: Seq[String], serverTimeZone: Option[ZoneId]): String = null override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = null diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDatabricksSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDatabricksSuite.scala index b3e85b873..d402c051e 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDatabricksSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDatabricksSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -170,21 +171,21 @@ class SqlGeneratorDatabricksSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala index ff354bd67..1fb61238d 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -169,21 +170,21 @@ class SqlGeneratorDb2Suite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > TIMESTAMP('1970-01-21 01:56:01.000')") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDenodoSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDenodoSuite.scala index d69dd0784..53039eafa 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDenodoSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDenodoSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -175,21 +176,21 @@ class SqlGeneratorDenodoSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > TIMESTAMP '1970-01-21 01:56:01.000+02:00'") } "return the correct condition for string offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorGenericSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorGenericSuite.scala index 46ef8b1f7..6a46c4ffc 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorGenericSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorGenericSuite.scala @@ -22,8 +22,9 @@ import za.co.absa.pramen.api.sql.SqlGeneratorBase.MAX_STRING_OFFSET_CHARACTERS import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory import za.co.absa.pramen.core.sql.SqlGeneratorGeneric +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz -import java.time.{Instant, LocalDate} +import java.time.{Instant, LocalDate, ZoneId} class SqlGeneratorGenericSuite extends AnyWordSpec { @@ -213,62 +214,62 @@ class SqlGeneratorGenericSuite extends AnyWordSpec { "getDataQueryIncremental" should { "info date is absent" when { "work without offsets" in { - val sql = genDate.getDataQueryIncremental("table1", None, None, None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1") } "work with only from offset" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE offset > 1") } "work with only to offset" in { - val sql = genDate.getDataQueryIncremental("table1", None, None, Some(OffsetValue.IntegralValue(1)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, None, Some(OffsetValue.IntegralValue(1)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE offset <= 1") } "work with from and to offsets" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE offset >= 1 AND offset <= 2") } "work with from and to offsets, column projection" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns, None) assert(sql == "SELECT A, D, \"Column with spaces\" FROM table1 WHERE offset >= 1 AND offset <= 2") } } "info date is present" when { "work without offsets" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE D = date'2020-08-17'") } "work with only from offset" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE D = date'2020-08-17' AND offset > 1") } "work with only to offset" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.IntegralValue(1)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.IntegralValue(1)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE D = date'2020-08-17' AND offset <= 1") } "work with from and to offsets" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WHERE D = date'2020-08-17' AND offset >= 1 AND offset <= 2") } "work with from and to offsets, column projection" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns, None) assert(sql == "SELECT A, D, \"Column with spaces\" FROM table1 WHERE D = date'2020-08-17' AND offset >= 1 AND offset <= 2") } @@ -400,21 +401,21 @@ class SqlGeneratorGenericSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } @@ -425,7 +426,7 @@ class SqlGeneratorGenericSuite extends AnyWordSpec { SqlGeneratorBase.validateOffsetValue(OffsetValue.IntegralValue(1)) } - "just return for a conforming string offset balue" in { + "just return for a conforming string offset value" in { SqlGeneratorBase.validateOffsetValue(OffsetValue.StringValue("AAA")) } @@ -443,3 +444,7 @@ class SqlGeneratorGenericSuite extends AnyWordSpec { } } } + +object SqlGeneratorGenericSuite { + val testDefaultTz: ZoneId = ZoneId.of("Africa/Johannesburg") +} diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHiveSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHiveSuite.scala index b46085955..0b2c3adee 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHiveSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHiveSuite.scala @@ -21,6 +21,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory import za.co.absa.pramen.core.sql.SqlGeneratorHive +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -203,21 +204,21 @@ class SqlGeneratorHiveSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHsqlSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHsqlSuite.scala index ac0da5ca6..526b10bc0 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHsqlSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorHsqlSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -174,21 +175,21 @@ class SqlGeneratorHsqlSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > TIMESTAMP '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMicrosoftSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMicrosoftSuite.scala index 1d4028ba5..6cc82e295 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMicrosoftSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMicrosoftSuite.scala @@ -21,6 +21,7 @@ import za.co.absa.pramen.api.offset.{OffsetInfo, OffsetType, OffsetValue} import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory import za.co.absa.pramen.core.sql.SqlGeneratorMicrosoft +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -327,62 +328,62 @@ class SqlGeneratorMicrosoftSuite extends AnyWordSpec { "integral offset" when { "info date is absent" when { "work without offsets" in { - val sql = genDate.getDataQueryIncremental("table1", None, None, None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK)") } "work with only from offset" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset > 1") } "work with only to offset" in { - val sql = genDate.getDataQueryIncremental("table1", None, None, Some(OffsetValue.IntegralValue(1)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, None, Some(OffsetValue.IntegralValue(1)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset <= 1") } "work with from and to offsets" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset >= 1 AND offset <= 2") } "work with from and to offsets, column projection" in { - val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns) + val sql = genDate.getDataQueryIncremental("table1", None, Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns, None) assert(sql == "SELECT A, D, [Column with spaces] FROM table1 WITH (NOLOCK) WHERE offset >= 1 AND offset <= 2") } } "info date is present" when { "work without offsets" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE D = CONVERT(DATE, '2020-08-17', 23)") } "work with only from offset" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), None, Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE D = CONVERT(DATE, '2020-08-17', 23) AND offset > 1") } "work with only to offset" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.IntegralValue(1)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.IntegralValue(1)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE D = CONVERT(DATE, '2020-08-17', 23) AND offset <= 1") } "work with from and to offsets" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE D = CONVERT(DATE, '2020-08-17', 23) AND offset >= 1 AND offset <= 2") } "work with from and to offsets, column projection" in { - val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns) + val sql = genDate.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.IntegralValue(1)), Some(OffsetValue.IntegralValue(2)), columns, None) assert(sql == "SELECT A, D, [Column with spaces] FROM table1 WITH (NOLOCK) WHERE D = CONVERT(DATE, '2020-08-17', 23) AND offset >= 1 AND offset <= 2") } @@ -393,62 +394,62 @@ class SqlGeneratorMicrosoftSuite extends AnyWordSpec { val offset2 = Instant.ofEpochMilli(1727762000) "info date is absent" when { "work without offsets" in { - val sql = genDateTime.getDataQueryIncremental("table1", None, None, None, Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", None, None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK)") } "work with only from offset" in { - val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), None, Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), None, Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset > '1970-01-21 01:56:01.000'") } "work with only to offset" in { - val sql = genDateTime.getDataQueryIncremental("table1", None, None, Some(OffsetValue.DateTimeValue(offset1)), Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", None, None, Some(OffsetValue.DateTimeValue(offset1)), Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset <= '1970-01-21 01:56:01.000'") } "work with from and to offsets" in { - val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE offset >= '1970-01-21 01:56:01.000' AND offset <= '1970-01-21 01:56:02.000'") } "work with from and to offsets, column projection" in { - val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), columns) + val sql = genDateTime.getDataQueryIncremental("table1", None, Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), columns, Some(testDefaultTz)) assert(sql == "SELECT A, D, [Column with spaces] FROM table1 WITH (NOLOCK) WHERE offset >= '1970-01-21 01:56:01.000' AND offset <= '1970-01-21 01:56:02.000'") } } "info date is present" when { "work without offsets" in { - val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), None, None, Seq.empty, None) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE CONVERT(DATE, D, 23) = CONVERT(DATE, '2020-08-17', 23)") } "work with only from offset" in { - val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), None, Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), None, Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE CONVERT(DATE, D, 23) = CONVERT(DATE, '2020-08-17', 23) AND offset > '1970-01-21 01:56:01.000'") } "work with only to offset" in { - val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.DateTimeValue(offset1)), Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), None, Some(OffsetValue.DateTimeValue(offset1)), Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE CONVERT(DATE, D, 23) = CONVERT(DATE, '2020-08-17', 23) AND offset <= '1970-01-21 01:56:01.000'") } "work with from and to offsets" in { - val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), Seq.empty) + val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), Seq.empty, Some(testDefaultTz)) assert(sql == "SELECT * FROM table1 WITH (NOLOCK) WHERE CONVERT(DATE, D, 23) = CONVERT(DATE, '2020-08-17', 23) AND offset >= '1970-01-21 01:56:01.000' AND offset <= '1970-01-21 01:56:02.000'") } "work with from and to offsets, column projection" in { - val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), columns) + val sql = genDateTime.getDataQueryIncremental("table1", Some(date1), Some(OffsetValue.DateTimeValue(offset1)), Some(OffsetValue.DateTimeValue(offset2)), columns, Some(testDefaultTz)) assert(sql == "SELECT A, D, [Column with spaces] FROM table1 WITH (NOLOCK) WHERE CONVERT(DATE, D, 23) = CONVERT(DATE, '2020-08-17', 23) AND offset >= '1970-01-21 01:56:01.000' AND offset <= '1970-01-21 01:56:02.000'") } @@ -459,21 +460,21 @@ class SqlGeneratorMicrosoftSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMySqlSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMySqlSuite.scala index bb7560bcf..cb6a45abd 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMySqlSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorMySqlSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -170,21 +171,21 @@ class SqlGeneratorMySqlSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorOracleSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorOracleSuite.scala index d185ebba0..35107550f 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorOracleSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorOracleSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -163,21 +164,21 @@ class SqlGeneratorOracleSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > TIMESTAMP '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorPostgreSqlSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorPostgreSqlSuite.scala index 4693c3660..02431e883 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorPostgreSqlSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorPostgreSqlSuite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -170,21 +171,21 @@ class SqlGeneratorPostgreSqlSuite extends AnyWordSpec { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '1970-01-21 01:56:01.000'") } "return the correct condition for string offsets" in { val actual = genDate.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorSasSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorSasSuite.scala index 09896c25b..f476ff113 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorSasSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorSasSuite.scala @@ -22,6 +22,7 @@ import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, Sq import za.co.absa.pramen.core.fixtures.RelationalDbFixture import za.co.absa.pramen.core.mocks.DummySqlConfigFactory import za.co.absa.pramen.core.samples.RdbExampleTable +import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.sql.Connection import java.time.{Instant, LocalDate} @@ -192,21 +193,21 @@ class SqlGeneratorSasSuite extends AnyWordSpec with RelationalDbFixture { "getOffsetWhereCondition" should { "return the correct condition for integral offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1)) + .getOffsetWhereCondition("offset", "<", OffsetValue.IntegralValue(1), None) assert(actual == "offset < 1") } "return the correct condition for datetime offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000))) + .getOffsetWhereCondition("offset", ">", OffsetValue.DateTimeValue(Instant.ofEpochMilli(1727761000)), Some(testDefaultTz)) assert(actual == "offset > '21Jan1970:01:56:01.000'dt") } "return the correct condition for string offsets" in { val actual = gen.asInstanceOf[SqlGeneratorBase] - .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA")) + .getOffsetWhereCondition("offset", ">=", OffsetValue.StringValue("AAA"), None) assert(actual == "offset >= 'AAA'") } From 9400b899538434d2e6a1e849a766babb12bf66c6 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Mon, 19 Oct 2026 08:58:28 +0200 Subject: [PATCH 2/4] Add logging of timestamp timezone conversion for datetime offsets and extract common conversion logic into SqlGeneratorBase --- .../za/co/absa/pramen/api/sql/SqlGeneratorBase.scala | 12 +++++++++++- .../pramen/core/sql/SqlGeneratorDatabricks.scala | 6 ++---- .../za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala | 6 ++---- .../co/absa/pramen/core/sql/SqlGeneratorDenodo.scala | 1 + .../absa/pramen/core/sql/SqlGeneratorGeneric.scala | 6 ++---- .../co/absa/pramen/core/sql/SqlGeneratorHive.scala | 6 ++---- .../co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala | 6 ++---- .../absa/pramen/core/sql/SqlGeneratorMicrosoft.scala | 3 +++ .../co/absa/pramen/core/sql/SqlGeneratorMySQL.scala | 6 ++---- .../co/absa/pramen/core/sql/SqlGeneratorOracle.scala | 6 ++---- .../pramen/core/sql/SqlGeneratorPostgreSQL.scala | 7 ++----- .../za/co/absa/pramen/core/sql/SqlGeneratorSas.scala | 1 + 12 files changed, 32 insertions(+), 34 deletions(-) diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala index 0bb2a9d4c..6fd1aa448 100644 --- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala +++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala @@ -16,11 +16,12 @@ package za.co.absa.pramen.api.sql +import org.slf4j.{Logger, LoggerFactory} import za.co.absa.pramen.api.offset.{OffsetInfo, OffsetValue} import java.sql.Connection import java.time.format.DateTimeFormatter -import java.time.{LocalDate, ZoneId} +import java.time.{Instant, LocalDate, LocalDateTime, ZoneId} import scala.collection.mutable.ListBuffer import scala.util.Try @@ -33,6 +34,7 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { import SqlGeneratorBase._ protected val timestampGenericDbFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS") + protected val log: Logger = LoggerFactory.getLogger(this.getClass) /** * This returns characters used for escaping into a mode that allows special characters in identifiers. @@ -257,6 +259,14 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { final protected def infoDateColumn: String = { escape(sqlConfig.infoDateColumn) } + + final protected def getGenericTimestampString(ts: Instant, serverTimeZone: Option[ZoneId]): String = { + val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) + val ldt = LocalDateTime.ofInstant(ts, tz) + val tsLiteral = timestampGenericDbFormatter.format(ldt) + log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") + tsLiteral + } } object SqlGeneratorBase { diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala index 51e2a6b5a..9cfd7f420 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala @@ -23,7 +23,7 @@ import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.DatabricksDialect import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} object SqlGeneratorDatabricks { private val log = LoggerFactory.getLogger(this.getClass) @@ -113,9 +113,7 @@ class SqlGeneratorDatabricks(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala index b3a271eaa..bad06bac1 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -99,9 +99,7 @@ class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition TIMESTAMP('$tsLiteral')" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala index f7788d0e1..a6688e4e5 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala @@ -115,6 +115,7 @@ class SqlGeneratorDenodo(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val zdt = ZonedDateTime.ofInstant(ts, tz) val tsLiteral = timestampDbFormatter.format(zdt) + log.info(s"Denodo: Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala index 6007f2afa..416faaee6 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorGeneric.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorGeneric(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -99,9 +99,7 @@ class SqlGeneratorGeneric(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConf override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala index b27c31c24..548371504 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala @@ -23,7 +23,7 @@ import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.HiveDialect import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} object SqlGeneratorHive { private val log = LoggerFactory.getLogger(this.getClass) @@ -113,9 +113,7 @@ class SqlGeneratorHive(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala index 6b360a645..b67954b2d 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHsqlDb.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorHsqlDb(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -99,9 +99,7 @@ class SqlGeneratorHsqlDb(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala index 9aa23bc59..872c2dac1 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala @@ -16,6 +16,7 @@ package za.co.absa.pramen.core.sql +import org.slf4j.LoggerFactory import za.co.absa.pramen.api.offset.{OffsetInfo, OffsetValue} import za.co.absa.pramen.api.sql.SqlGeneratorBase.{needsEscaping, validateIdentifier, validateOffsetValue} import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGenerator} @@ -31,6 +32,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) private val timestampMsDbFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS") private val isIso = sqlConfig.dateFormatApp.toLowerCase.startsWith("yyyy-mm-dd") + private val log = LoggerFactory.getLogger(this.getClass) // 23 is "yyyy-MM-dd", see https://www.mssqltips.com/sqlservertip/1145/date-and-time-conversions-using-sql-server/ private val isoFormatMsSqlRef = 23 @@ -227,6 +229,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampMsDbFormatter.format(ldt) + log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala index 3181a142e..874b2e281 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -95,9 +95,7 @@ class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala index a13e18835..b6102fed9 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorOracle.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorOracle(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -92,9 +92,7 @@ class SqlGeneratorOracle(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala index 0a5dab1b5..34ae25242 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorPostgreSQL.scala @@ -20,7 +20,7 @@ import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, ZoneId} class SqlGeneratorPostgreSQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -91,13 +91,10 @@ class SqlGeneratorPostgreSQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC } } - override def getOffsetWhereCondition(column: String, condition: String, offset: OffsetValue, serverTimeZone: Option[ZoneId]): String = { offset match { case OffsetValue.DateTimeValue(ts) => - val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) - val ldt = LocalDateTime.ofInstant(ts, tz) - val tsLiteral = timestampGenericDbFormatter.format(ldt) + val tsLiteral = getGenericTimestampString(ts, serverTimeZone) s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala index bed31e3cd..051688425 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala @@ -197,6 +197,7 @@ class SqlGeneratorSas(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampSasDbFormatter.format(ldt) + log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") s"$column $condition '$tsLiteral'dt" case OffsetValue.IntegralValue(value) => s"$column $condition $value" From 4a36afc7c5f758141e6403d4a88d57918f222980 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Mon, 19 Oct 2026 16:11:02 +0200 Subject: [PATCH 3/4] Add server timezone detection for Hive and Denodo by parsing current timestamp and epoch seconds, and log epoch millis in timestamp conversions --- .../pramen/api/sql/SqlGeneratorBase.scala | 4 +- .../pramen/core/sql/SqlGeneratorDenodo.scala | 40 +++- .../pramen/core/sql/SqlGeneratorHive.scala | 38 ++++ .../core/sql/SqlGeneratorMicrosoft.scala | 2 +- .../pramen/core/sql/SqlGeneratorSas.scala | 2 +- .../co/absa/pramen/core/utils/TimeUtils.scala | 45 +++- .../core/tests/utils/TimeUtilsSuite.scala | 213 ++++++++++++++++++ 7 files changed, 337 insertions(+), 7 deletions(-) diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala index 6fd1aa448..7b067200f 100644 --- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala +++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala @@ -81,8 +81,6 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` * against the given connection and parsing the first column of the first row as a timezone identifier. * - * The statement and result set used to run the query are always closed, regardless of the outcome. - * * @param connection An open JDBC connection to the database server whose timezone should be determined. * @return A `Success` containing the server's `ZoneId` if the query returns a valid timezone identifier, * or a `Failure` if the query fails, returns no rows, returns NULL, or returns a value that cannot @@ -264,7 +262,7 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator { val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampGenericDbFormatter.format(ldt) - log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") + log.info(s"Timestamp conversion for timezone '$tz': ${ts.toEpochMilli}: $ts -> '$tsLiteral'") tsLiteral } } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala index a6688e4e5..4b4cc6394 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDenodo.scala @@ -21,9 +21,12 @@ import org.slf4j.LoggerFactory import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.DenodoDialect +import za.co.absa.pramen.core.utils.TimeUtils +import java.sql.Connection import java.time.format.DateTimeFormatter import java.time.{LocalDate, LocalTime, ZoneId, ZonedDateTime} +import scala.util.Try object SqlGeneratorDenodo { private val log = LoggerFactory.getLogger(this.getClass) @@ -78,6 +81,41 @@ class SqlGeneratorDenodo(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi override def getServerTimezoneQuery: String = "SELECT CURRENT_TIMESTAMP AS ts, GETTIMEINMILLIS(CURRENT_TIMESTAMP) / 1000 AS epoch_sec" + /** + * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` + * against the given connection and parsing the offset of the returned timestamp. + * + * @param connection An open JDBC connection to the database server whose timezone should be determined. + * @return A `Success` containing the server's `ZoneId` if the query returns a valid timezone identifier, + * or a `Failure` if the query fails, returns no rows, returns NULL, or returns a value that cannot + * be parsed as a `ZoneId`. + */ + override def getServerTimezone(connection: Connection): Try[ZoneId] = { + Try { + val query = getServerTimezoneQuery + val statement = connection.createStatement() + try { + val resultSet = statement.executeQuery(query) + try { + if (resultSet.next()) { + val currentTimestamp = resultSet.getString(1) + if (currentTimestamp == null) { + throw new IllegalStateException(s"The server timezone query returned NULL: $query") + } + TimeUtils.getTimezoneFromTimestampString(currentTimestamp).get + } else { + throw new IllegalStateException(s"The server timezone query returned no rows: $query") + } + } finally { + resultSet.close() + } + } finally { + statement.close() + } + } + } + + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { if (sqlConfig.infoDateType == SqlColumnType.DATE && dateBegin == dateEnd) { val dateBeginLit = getDateLiteral(dateBegin) @@ -115,7 +153,7 @@ class SqlGeneratorDenodo(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfi val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val zdt = ZonedDateTime.ofInstant(ts, tz) val tsLiteral = timestampDbFormatter.format(zdt) - log.info(s"Denodo: Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") + log.info(s"Denodo: Timestamp conversion for timezone '$tz': ${ts.toEpochMilli}: $ts -> '$tsLiteral'") s"$column $condition TIMESTAMP '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala index 548371504..5dd4cbb87 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorHive.scala @@ -21,9 +21,12 @@ import org.slf4j.LoggerFactory import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} import za.co.absa.pramen.core.sql.dialects.HiveDialect +import za.co.absa.pramen.core.utils.TimeUtils +import java.sql.Connection import java.time.format.DateTimeFormatter import java.time.{LocalDate, ZoneId} +import scala.util.Try object SqlGeneratorHive { private val log = LoggerFactory.getLogger(this.getClass) @@ -73,6 +76,41 @@ class SqlGeneratorHive(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) override def getServerTimezoneQuery: String = "SELECT current_timestamp AS ts, unix_timestamp() AS epoch_sec" + /** + * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` + * against the given connection and parsing both the current timestamp as string and as a epoch seconds. + * + * @param connection An open JDBC connection to the database server whose timezone should be determined. + * @return A `Success` containing the server's `ZoneId` if the query returns a valid timezone identifier, + * or a `Failure` if the query fails, returns no rows, returns NULL, or returns a value that cannot + * be parsed as a `ZoneId`. + */ + override def getServerTimezone(connection: Connection): Try[ZoneId] = { + Try { + val query = getServerTimezoneQuery + val statement = connection.createStatement() + try { + val resultSet = statement.executeQuery(query) + try { + if (resultSet.next()) { + val currentTimestamp = resultSet.getString(1) + val currentEpochSeconds = resultSet.getLong(2) + if (currentTimestamp == null) { + throw new IllegalStateException(s"The server timezone query returned NULL: $query") + } + TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds(currentTimestamp, currentEpochSeconds).get + } else { + throw new IllegalStateException(s"The server timezone query returned no rows: $query") + } + } finally { + resultSet.close() + } + } finally { + statement.close() + } + } + } + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala index 872c2dac1..4c30d0703 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala @@ -229,7 +229,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampMsDbFormatter.format(ldt) - log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") + log.info(s"Timestamp conversion for timezone '$tz': ${ts.toEpochMilli}: $ts -> '$tsLiteral'") s"$column $condition '$tsLiteral'" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala index 051688425..4dbba51a5 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorSas.scala @@ -197,7 +197,7 @@ class SqlGeneratorSas(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) val tz = serverTimeZone.getOrElse(throw new IllegalArgumentException(s"Server time zone must be provided when offsets have 'datetime' type")) val ldt = LocalDateTime.ofInstant(ts, tz) val tsLiteral = timestampSasDbFormatter.format(ldt) - log.info(s"Timestamp conversion for timezone '$tz': $ts -> '$tsLiteral'") + log.info(s"Timestamp conversion for timezone '$tz': ${ts.toEpochMilli}: $ts -> '$tsLiteral'") s"$column $condition '$tsLiteral'dt" case OffsetValue.IntegralValue(value) => s"$column $condition $value" diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala index 00cce0626..e88ce148e 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala @@ -18,8 +18,10 @@ package za.co.absa.pramen.core.utils import org.slf4j.LoggerFactory -import java.time.{Duration, Instant} +import java.time.format.DateTimeFormatter +import java.time._ import java.util.Locale +import scala.util.Try object TimeUtils { import scala.math.Integral.Implicits._ @@ -160,8 +162,49 @@ object TimeUtils { log.info(s"$description $elapsedTimeStr") } + /** + * Extracts the timezone offset from a timestamp string and returns it as a ZoneId. + * + * Example input: "2026-10-16 15:02:28.436 +0200" will return ZoneId "+02:00". + * The milliseconds part is optional. + * + * @param tsStr the timestamp string containing a timezone offset. + * @return the zone offset as a ZoneId, or a failure if the string cannot be parsed. + */ + def getTimezoneFromTimestampString(tsStr: String): Try[ZoneId] = { + Try { + val offsetDateTime = OffsetDateTime.parse(tsStr.trim, timestampWithOffsetFormatter) + offsetDateTime.getOffset: ZoneId + } + } + + /** + * Extracts the timezone offset from a timestamp string and returns it as a ZoneId. + * + * Example input: "2026-10-16 14:56:28.038" and 1792155388 will return ZoneId "+02:00". + * The milliseconds part is optional. + * The calculated offset is rounded to the nearest 5 minutes. + * + * @param tsStr the timestamp string containing a timezone offset. + * @return the zone offset as a ZoneId, or a failure if the string cannot be parsed. + */ + def getTimezoneFromTimestampStringAndEpochSeconds(tsStr: String, epochSeconds: Long): Try[ZoneId] = { + Try { + val localDateTime = LocalDateTime.parse(tsStr.trim, timestampWithoutOffsetFormatter) + val localEpochSeconds = localDateTime.toEpochSecond(ZoneOffset.UTC) + val diffSeconds = localEpochSeconds - epochSeconds + val roundingSeconds = 300L + val offsetSeconds = Math.round(diffSeconds.toDouble / roundingSeconds.toDouble) * roundingSeconds + + ZoneOffset.ofTotalSeconds(offsetSeconds.toInt): ZoneId + } + } + private val millisecondsPerSecond: Long = 1000 private val millisecondsPerMinute: Long = millisecondsPerSecond * 60 private val millisecondsPerHour: Long = millisecondsPerMinute * 60 private val millisecondsPerDay: Long = millisecondsPerHour * 24 + + private val timestampWithOffsetFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss[.SSS] Z", Locale.US) + private val timestampWithoutOffsetFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss[.SSS]", Locale.US) } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala index 975ee6732..d41a5aefc 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala @@ -126,4 +126,217 @@ class TimeUtilsSuite extends AnyWordSpec { } } + "getTimezoneFromTimestampString" should { + "extract a positive offset from a timestamp with milliseconds" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28.436 +0200") + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + assert(result.get.toString == "+02:00") + } + + "extract a positive offset from a timestamp without milliseconds" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28 +0200") + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + assert(result.get.toString == "+02:00") + } + + "extract a negative offset with non-zero minutes" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28.436 -0530") + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHoursMinutes(-5, -30)) + assert(result.get.toString == "-05:30") + } + + "extract a zero offset as UTC" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28.436 +0000") + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.UTC) + assert(result.get.toString == "Z") + } + + "ignore leading and trailing whitespace" in { + val result = TimeUtils.getTimezoneFromTimestampString(" 2026-10-16 15:02:28.436 +0200 ") + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + } + + "return the same offset regardless of the date and time part" in { + val result1 = TimeUtils.getTimezoneFromTimestampString("2020-01-01 00:00:00 +0300") + val result2 = TimeUtils.getTimezoneFromTimestampString("2025-12-31 23:59:59.999 +0300") + + assert(result1.isSuccess) + assert(result2.isSuccess) + assert(result1.get == result2.get) + assert(result1.get == java.time.ZoneOffset.ofHours(3)) + } + + "return a failure for an empty string" in { + val result = TimeUtils.getTimezoneFromTimestampString("") + + assert(result.isFailure) + } + + "return a failure for a blank string" in { + val result = TimeUtils.getTimezoneFromTimestampString(" ") + + assert(result.isFailure) + } + + "return a failure when the offset is missing" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28.436") + + assert(result.isFailure) + } + + "return a failure when only the date is provided" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16") + + assert(result.isFailure) + } + + "return a failure for an invalid date" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-13-45 15:02:28.436 +0200") + + assert(result.isFailure) + } + + "return a failure for an out of range offset" in { + val result = TimeUtils.getTimezoneFromTimestampString("2026-10-16 15:02:28.436 +2500") + + assert(result.isFailure) + } + + "return a failure for a completely invalid string" in { + val result = TimeUtils.getTimezoneFromTimestampString("not a timestamp") + + assert(result.isFailure) + } + } + + "getTimezoneFromTimestampStringAndEpochSeconds" should { + // "2026-10-16 14:56:28" interpreted as UTC corresponds to 1792162588 epoch seconds + val localAsUtcEpochSeconds = 1792162588L + + "extract a positive offset from a timestamp with milliseconds" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038", 1792155388L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + assert(result.get.toString == "+02:00") + } + + "extract a positive offset from a timestamp without milliseconds" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", 1792155388L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + assert(result.get.toString == "+02:00") + } + + "extract a zero offset as UTC" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038", localAsUtcEpochSeconds) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.UTC) + assert(result.get.toString == "Z") + } + + "extract a negative offset with non-zero minutes" in { + val epochSeconds = localAsUtcEpochSeconds + 5 * 3600 + 30 * 60 + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038", epochSeconds) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHoursMinutes(-5, -30)) + assert(result.get.toString == "-05:30") + } + + "extract a negative whole-hour offset" in { + val epochSeconds = localAsUtcEpochSeconds + 10 * 3600 + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", epochSeconds) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(-10)) + assert(result.get.toString == "-10:00") + } + + "round the offset to the nearest minute when epoch seconds is slightly ahead" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038", 1792155388L + 20L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + } + + "round the offset to the nearest minute when epoch seconds is slightly behind" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038", 1792155388L - 25L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + } + + "round the offset down to the nearest 5 minutes" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", 1792155388L - 40L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHoursMinutes(2, 0)) + assert(result.get.toString == "+02:00") + } + + "round the offset up to the next 5 minutes when the difference is more than half of 5 minutes" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", 1792155388L - 200L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHoursMinutes(2, 5)) + assert(result.get.toString == "+02:05") + } + + "ignore leading and trailing whitespace" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds(" 2026-10-16 14:56:28.038 ", 1792155388L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + } + + "return a failure for an empty string" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("", 1792155388L) + + assert(result.isFailure) + } + + "return a failure for a blank string" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds(" ", 1792155388L) + + assert(result.isFailure) + } + + "return a failure when only the date is provided" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16", 1792155388L) + + assert(result.isFailure) + } + + "return a failure for an invalid date" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-13-45 14:56:28.038", 1792155388L) + + assert(result.isFailure) + } + + "return a failure for a completely invalid string" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("not a timestamp", 1792155388L) + + assert(result.isFailure) + } + + "return a failure when the computed offset is out of range" in { + val epochSeconds = localAsUtcEpochSeconds - 20 * 3600 + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", epochSeconds) + + assert(result.isFailure) + } + } } From 9a457e7c269af295e6f03c5548fed77fdce17fb4 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Tue, 20 Oct 2026 14:45:40 +0200 Subject: [PATCH 4/4] Fix server timezone detection queries for Databricks, MySQL, DB2, and MS SQL Server, use the detected timezone in JDBC readers, and support variable-length fractional seconds when parsing timestamps --- .../core/reader/TableReaderJdbcBase.scala | 2 +- .../core/sql/SqlGeneratorDatabricks.scala | 2 +- .../pramen/core/sql/SqlGeneratorDb2.scala | 55 +++++++++- .../core/sql/SqlGeneratorMicrosoft.scala | 11 +- .../pramen/core/sql/SqlGeneratorMySQL.scala | 3 +- .../co/absa/pramen/core/utils/TimeUtils.scala | 15 ++- .../core/tests/sql/SqlGeneratorDb2Suite.scala | 103 ++++++++++++++++++ .../core/tests/utils/TimeUtilsSuite.scala | 8 ++ 8 files changed, 185 insertions(+), 14 deletions(-) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala index 40ccb7caa..a653c98ff 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcBase.scala @@ -73,7 +73,7 @@ abstract class TableReaderJdbcBase(jdbcReaderConfig: TableReaderJdbcConfig, jdbcReaderConfig.infoDateType, jdbcReaderConfig.infoDateFormat, jdbcReaderConfig.offsetInfoOpt, - jdbcReaderConfig.defaultTimeZone, + jdbcReaderConfig.serverTimeZone.getOrElse(jdbcReaderConfig.defaultTimeZone), jdbcReaderConfig.identifierQuotingPolicy, jdbcReaderConfig.sqlGeneratorClass, ConfigUtils.getExtraConfig(conf, "sql")) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala index 9cfd7f420..11e48058a 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDatabricks.scala @@ -71,7 +71,7 @@ class SqlGeneratorDatabricks(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlC s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } - override def getServerTimezoneQuery: String = "SET spark.sql.session.timeZone" + override def getServerTimezoneQuery: String = "SELECT current_timezone()" override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala index bad06bac1..ce88fe035 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorDb2.scala @@ -19,8 +19,10 @@ package za.co.absa.pramen.core.sql import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase} +import java.sql.Connection import java.time.format.DateTimeFormatter -import java.time.{LocalDate, ZoneId} +import java.time.{LocalDate, ZoneId, ZoneOffset} +import scala.util.Try class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) { private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp) @@ -59,6 +61,28 @@ class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) override def getServerTimezoneQuery: String = "SELECT CURRENT TIMEZONE FROM SYSIBM.SYSDUMMY1" + override def getServerTimezone(connection: Connection): Try[ZoneId] = { + Try { + val query = getServerTimezoneQuery + val statement = connection.createStatement() + try { + val resultSet = statement.executeQuery(query) + try { + if (resultSet.next()) { + val timezoneRaw = resultSet.getBigDecimal(1).intValue() + SqlGeneratorDb2.convertDb2TzOffsets(timezoneRaw) + } else { + throw new IllegalStateException(s"The server timezone query returned no rows: $query") + } + } finally { + resultSet.close() + } + } finally { + statement.close() + } + } + } + override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { val dateBeginLit = getDateLiteral(dateBegin) val dateEndLit = getDateLiteral(dateEnd) @@ -114,3 +138,32 @@ class SqlGeneratorDb2(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) limit.map(n => s" LIMIT $n").getOrElse("") } } + +object SqlGeneratorDb2 { + /** + * Converts a DB2 timezone offset value into a Java `ZoneId`. + * + * DB2 returns timezone as ±hhmmss + * Examples: + * 20000 -> +02:00:00 + * -53000 -> -05:30:00 + * + * @param db2Offset the timezone offset as returned by DB2, encoded as a signed integer in the `hhmmss` format + * @return a `ZoneOffset` (as a `ZoneId`) representing the same offset from UTC + * @throws java.time.DateTimeException if the decoded hours, minutes or seconds are outside the range supported by `ZoneOffset` + */ + def convertDb2TzOffsets(db2Offset: Int): ZoneId = { + val sign = if (db2Offset < 0) -1 else 1 + val abs = math.abs(db2Offset) + + val hours = abs / 10000 + val minutes = (abs % 10000) / 100 + val seconds = abs % 100 + + ZoneOffset.ofHoursMinutesSeconds( + sign * hours, + sign * minutes, + sign * seconds + ) + } +} \ No newline at end of file diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala index 4c30d0703..36fd6ee3f 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala @@ -24,7 +24,7 @@ import za.co.absa.pramen.core.utils.MutableStack import java.sql.Connection import java.time.format.DateTimeFormatter -import java.time.{LocalDate, LocalDateTime, ZoneId} +import java.time.{LocalDate, LocalDateTime, ZoneId, ZoneOffset} import scala.collection.mutable.ListBuffer import scala.util.Try @@ -76,7 +76,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { s"SELECT ${getLimit(limit)}${columnExpr(columns)} FROM ${escape(tableName)} WITH (NOLOCK) WHERE $where" } - def getServerTimezoneQuery: String = "SELECT CURRENT_TIMEZONE()" + def getServerTimezoneQuery: String = "SELECT DATEPART(TZOFFSET, SYSDATETIMEOFFSET())" /** * Determines the timezone of the database server by executing the query returned by `getServerTimezoneQuery` @@ -97,11 +97,8 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator { val resultSet = statement.executeQuery(query) try { if (resultSet.next()) { - val timezoneStr = resultSet.getString(1) - if (timezoneStr == null) { - throw new IllegalStateException(s"The server timezone query returned NULL: $query") - } - ZoneId.of(timezoneStr.trim) + val timezoneMinutes = resultSet.getString(1).toInt + ZoneOffset.ofTotalSeconds(timezoneMinutes * 60): ZoneId } else { throw new IllegalStateException(s"The server timezone query returned no rows: $query") } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala index 874b2e281..b4b833656 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMySQL.scala @@ -57,7 +57,8 @@ class SqlGeneratorMySQL(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig s"SELECT ${columnExpr(columns)} FROM ${escape(tableName)} WHERE $where${getLimit(limit)}" } - override def getServerTimezoneQuery: String = "SELECT @@session.time_zone" + override def getServerTimezoneQuery: String = + "SELECT CASE WHEN @@session.time_zone = 'SYSTEM' THEN @@system_time_zone ELSE @@session.time_zone END AS effective_time_zone" override def getWhere(dateBegin: LocalDate, dateEnd: LocalDate): String = { if (sqlConfig.infoDateType == SqlColumnType.DATETIME) { diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala index e88ce148e..9cdddb357 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/TimeUtils.scala @@ -18,8 +18,8 @@ package za.co.absa.pramen.core.utils import org.slf4j.LoggerFactory -import java.time.format.DateTimeFormatter import java.time._ +import java.time.format.DateTimeFormatter import java.util.Locale import scala.util.Try @@ -205,6 +205,15 @@ object TimeUtils { private val millisecondsPerHour: Long = millisecondsPerMinute * 60 private val millisecondsPerDay: Long = millisecondsPerHour * 24 - private val timestampWithOffsetFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss[.SSS] Z", Locale.US) - private val timestampWithoutOffsetFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss[.SSS]", Locale.US) + private val timestampWithoutOffsetFormatter: DateTimeFormatter = new java.time.format.DateTimeFormatterBuilder() + .appendPattern("yyyy-MM-dd HH:mm:ss") + .optionalStart() + .appendFraction(java.time.temporal.ChronoField.NANO_OF_SECOND, 0, 9, true) + .optionalEnd() + .toFormatter(Locale.US) + + private val timestampWithOffsetFormatter: DateTimeFormatter = new java.time.format.DateTimeFormatterBuilder() + .append(timestampWithoutOffsetFormatter) + .appendPattern(" Z") + .toFormatter(Locale.US) } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala index 1fb61238d..64534cce9 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorDb2Suite.scala @@ -20,6 +20,7 @@ import org.scalatest.wordspec.AnyWordSpec import za.co.absa.pramen.api.offset.OffsetValue import za.co.absa.pramen.api.sql.{QuotingPolicy, SqlColumnType, SqlGenerator, SqlGeneratorBase} import za.co.absa.pramen.core.mocks.DummySqlConfigFactory +import za.co.absa.pramen.core.sql.SqlGeneratorDb2._ import za.co.absa.pramen.core.tests.sql.SqlGeneratorGenericSuite.testDefaultTz import java.time.{Instant, LocalDate} @@ -141,6 +142,108 @@ class SqlGeneratorDb2Suite extends AnyWordSpec { assert(genDate.getCountQueryForSql("SELECT A FROM B") == "SELECT COUNT(*) FROM (SELECT A FROM B) AS query") } } + + // /** + // * Converts a DB2 timezone offset value into a Java `ZoneId`. + // * + // * DB2 returns timezone as ±hhmmss + // * Examples: + // * 20000 -> +02:00:00 + // * -53000 -> -05:30:00 + // * + // * @param db2Offset the timezone offset as returned by DB2, encoded as a signed integer in the `hhmmss` format + // * @return a `ZoneOffset` (as a `ZoneId`) representing the same offset from UTC + // * @throws java.time.DateTimeException if the decoded hours, minutes or seconds are outside the range supported by `ZoneOffset` + // */ + // def convertDb2TzOffsets(db2Offset: Int): ZoneId = { + // val sign = if (db2Offset < 0) -1 else 1 + // val abs = math.abs(db2Offset) + // + // val hours = abs / 10000 + // val minutes = (abs % 10000) / 100 + // val seconds = abs % 100 + // + // ZoneOffset.ofHoursMinutesSeconds( + // sign * hours, + // sign * minutes, + // sign * seconds + // ) + // } + "convertDb2TzOffsets" should { + "convert zero offset to UTC" in { + assert(convertDb2TzOffsets(0) == java.time.ZoneOffset.UTC) + } + + "convert positive whole hour offsets" in { + assert(convertDb2TzOffsets(20000) == java.time.ZoneOffset.ofHours(2)) + assert(convertDb2TzOffsets(10000) == java.time.ZoneOffset.ofHours(1)) + } + + "convert negative whole hour offsets" in { + assert(convertDb2TzOffsets(-50000) == java.time.ZoneOffset.ofHours(-5)) + assert(convertDb2TzOffsets(-120000) == java.time.ZoneOffset.ofHours(-12)) + } + + "convert positive offsets with minutes" in { + assert(convertDb2TzOffsets(53000) == java.time.ZoneOffset.ofHoursMinutes(5, 30)) + assert(convertDb2TzOffsets(54500) == java.time.ZoneOffset.ofHoursMinutes(5, 45)) + } + + "convert negative offsets with minutes" in { + assert(convertDb2TzOffsets(-53000) == java.time.ZoneOffset.ofHoursMinutes(-5, -30)) + assert(convertDb2TzOffsets(-33000) == java.time.ZoneOffset.ofHoursMinutes(-3, -30)) + } + + "convert offsets with seconds" in { + assert(convertDb2TzOffsets(53045) == java.time.ZoneOffset.ofHoursMinutesSeconds(5, 30, 45)) + assert(convertDb2TzOffsets(-53045) == java.time.ZoneOffset.ofHoursMinutesSeconds(-5, -30, -45)) + } + + "convert offsets smaller than one hour" in { + assert(convertDb2TzOffsets(3000) == java.time.ZoneOffset.ofHoursMinutes(0, 30)) + assert(convertDb2TzOffsets(-3000) == java.time.ZoneOffset.ofHoursMinutes(0, -30)) + assert(convertDb2TzOffsets(15) == java.time.ZoneOffset.ofTotalSeconds(15)) + assert(convertDb2TzOffsets(-15) == java.time.ZoneOffset.ofTotalSeconds(-15)) + } + + "convert boundary offsets" in { + assert(convertDb2TzOffsets(180000) == java.time.ZoneOffset.ofHours(18)) + assert(convertDb2TzOffsets(-180000) == java.time.ZoneOffset.ofHours(-18)) + } + + "return a zone id with the expected id string" in { + assert(convertDb2TzOffsets(20000).getId == "+02:00") + assert(convertDb2TzOffsets(-53000).getId == "-05:30") + assert(convertDb2TzOffsets(53045).getId == "+05:30:45") + } + + "throw an exception when hours are out of range" in { + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(190000) + } + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(-190000) + } + } + + "throw an exception when minutes are out of range" in { + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(6000) + } + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(-16000) + } + } + + "throw an exception when seconds are out of range" in { + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(60) + } + intercept[java.time.DateTimeException] { + convertDb2TzOffsets(-10099) + } + } + } "getDtable" should { "return the original table when a table is provided" in { diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala index d41a5aefc..b7f525442 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/TimeUtilsSuite.scala @@ -231,6 +231,14 @@ class TimeUtilsSuite extends AnyWordSpec { assert(result.get.toString == "+02:00") } + "extract a positive offset from a timestamp with milliseconds having 6 digits" in { + val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28.038123", 1792155388L) + + assert(result.isSuccess) + assert(result.get == java.time.ZoneOffset.ofHours(2)) + assert(result.get.toString == "+02:00") + } + "extract a positive offset from a timestamp without milliseconds" in { val result = TimeUtils.getTimezoneFromTimestampStringAndEpochSeconds("2026-10-16 14:56:28", 1792155388L)