Merge branch 'staging' into feat/postgres_ssl_ca

Author: vishalm0509

drivers/oracle/internal/incremental.go

@@ -3,11 +3,14 @@ package driver
 import (
 	"context"
 	"fmt"
+	"strings"
+	"time"
 
 	"github.com/datazip-inc/olake/constants"
 	"github.com/datazip-inc/olake/drivers/abstract"
 	"github.com/datazip-inc/olake/pkg/jdbc"
 	"github.com/datazip-inc/olake/types"
+	"github.com/datazip-inc/olake/utils/logger"
 )
 
 // StreamIncrementalChanges implements incremental sync for Oracle
@@ -43,9 +46,41 @@ func (o *Oracle) StreamIncrementalChanges(ctx context.Context, stream types.Stre
 }
 
 func (o *Oracle) FetchMaxCursorValues(ctx context.Context, stream types.StreamInterface) (any, any, error) {
-	maxPrimaryCursorValue, maxSecondaryCursorValue, err := jdbc.GetMaxCursorValues(ctx, o.client, constants.Oracle, stream)
+	maxPrimary, maxSecondary, err := jdbc.GetMaxCursorValues(ctx, o.client, constants.Oracle, stream)
 	if err != nil {
 		return nil, nil, err
 	}
-	return maxPrimaryCursorValue, maxSecondaryCursorValue, nil
+
+	primaryCursor, secondaryCursor := stream.Cursor()
+	maxPrimary = o.normalizeCursorValue(ctx, stream, primaryCursor, maxPrimary)
+	if secondaryCursor != "" {
+		maxSecondary = o.normalizeCursorValue(ctx, stream, secondaryCursor, maxSecondary)
+	}
+	return maxPrimary, maxSecondary, nil
+}
+
+// normalizeCursorTime converts a cursor time.Time from GetMaxCursorValues to plain UTC,
+// applying wall-clock strip for TZ-naive Oracle columns and .UTC() for TZ-aware ones.
+func (o *Oracle) normalizeCursorValue(ctx context.Context, stream types.StreamInterface, cursorField string, value any) any {
+	t, ok := value.(time.Time)
+	if !ok {
+		return value
+	}
+	var dataType string
+	query := jdbc.OracleColumnDataTypeQuery(stream.Namespace(), stream.Name(), strings.ToUpper(cursorField))
+	if err := o.client.QueryRowContext(ctx, query).Scan(&dataType); err != nil {
+		logger.Warnf("normalizeCursorValue: failed to get DATA_TYPE for %s.%s.%s, cursor may be incorrect: %s",
+			stream.Namespace(), stream.Name(), cursorField, err)
+		return value
+	}
+	// FetchMaxCursorValues bypasses dataTypeConverter, so go-ora attaches the session timezone
+	// to TZ-naive columns and the original timezone to TZ-aware columns. Normalize here so
+	// the cursor values entering FormatCursorValue are already plain UTC.
+	upper := strings.ToUpper(dataType)
+	if strings.Contains(upper, "WITH TIME ZONE") || strings.Contains(upper, "WITH LOCAL TIME ZONE") {
+		// TZ-aware: convert to the correct UTC instant.
+		return t.UTC()
+	}
+	// TZ-naive (TIMESTAMP, DATE): strip session-TZ offset, preserve wall-clock as UTC.
+	return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), time.UTC)
 }

drivers/oracle/internal/oracle.go

@@ -211,10 +211,33 @@ func (o *Oracle) ProduceSchema(ctx context.Context, streamName string) (*types.S
 	return stream, pkRows.Err()
 }
 
+// tzNaiveOracleTypes are Oracle column types that store only wall-clock time with no timezone.
+// go-ora attaches dbServerTimeZone when decoding all three via the same code path:
+//
+//	DATE (12)          → "DATE"         — Oracle DATE columns
+//	TimeStampDTY (180) → "TimeStampDTY" — TIMESTAMP columns in regular SELECT results
+//	TODO: Add support for TIMESTAMP (187)    → "TIMESTAMP"    — TIMESTAMP in RETURNING INTO / PL/SQL OUT params
+//
+// Strip the attached offset to preserve wall-clock digits consistently as UTC.
+var tzNaiveOracleTypes = map[string]bool{
+	"date":         true,
+	"timestampdty": true,
+}
+
 func (o *Oracle) dataTypeConverter(value interface{}, columnType string) (interface{}, error) {
 	if value == nil {
 		return nil, typeutils.ErrNullValue
 	}
 	olakeType := typeutils.ExtractAndMapColumnType(columnType, oracleTypeToDataTypes)
-	return typeutils.ReformatValue(olakeType, value)
+	result, err := typeutils.ReformatValue(olakeType, value)
+	if err != nil {
+		return result, err
+	}
+	// Strip the session-timezone offset that go-ora attaches to timezone-naive columns.
+	if tzNaiveOracleTypes[strings.ToLower(columnType)] {
+		if t, ok := result.(time.Time); ok {
+			result = time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), time.UTC)
+		}
+	}
+	return result, nil
 }

pkg/jdbc/jdbc.go

@@ -941,6 +941,11 @@ func OracleTableDetailsQuery(schemaName, tableName string) string {
 	return fmt.Sprintf("SELECT column_name, data_type, nullable, data_precision, data_scale FROM all_tab_columns WHERE owner = '%s' AND table_name = '%s'", schemaName, tableName)
 }
 
+// OracleColumnDataTypeQuery returns the query to fetch the data type of a column in OracleDB
+func OracleColumnDataTypeQuery(schemaName, tableName, columnName string) string {
+	return fmt.Sprintf("SELECT DATA_TYPE FROM ALL_TAB_COLUMNS WHERE OWNER = '%s' AND TABLE_NAME = '%s' AND COLUMN_NAME = '%s'", schemaName, tableName, columnName)
+}
+
 // OraclePrimaryKeyQuery returns the query to fetch all the primary key columns of a table in OracleDB
 func OraclePrimaryKeyColummsQuery(schemaName, tableName string) string {
 	return fmt.Sprintf(`SELECT cols.column_name FROM all_constraints cons, all_cons_columns cols WHERE cons.constraint_type = 'P' AND cons.constraint_name = cols.constraint_name AND cons.owner = cols.owner AND cons.owner = '%s' AND cols.table_name = '%s'`, schemaName, tableName)
@@ -1076,7 +1081,7 @@ func IncrementalValueFormatter(ctx context.Context, cursorField, argumentPlaceho
 	var dbDatatype string
 	switch opts.Driver {
 	case constants.Oracle:
-		query := fmt.Sprintf("SELECT DATA_TYPE FROM ALL_TAB_COLUMNS WHERE OWNER = '%s' AND TABLE_NAME = '%s' AND COLUMN_NAME = '%s'", stream.Namespace(), stream.Name(), cursorField)
+		query := OracleColumnDataTypeQuery(stream.Namespace(), stream.Name(), cursorField)
 		err = opts.Client.QueryRowContext(ctx, query).Scan(&dbDatatype)
 		if err != nil {
 			return "", nil, fmt.Errorf("failed to get column datatype: %s", err)