Add support for Fivetran's TIME_NAIVE data type

Author: amotl

CHANGES.md

@@ -9,6 +9,7 @@
 - Added support for `AlterTable::drop_columns` operation.
 - Added support for schema migrations.
 - Added support for history mode.
+- Added support for Fivetran's `TIME_NAIVE` data type.
 
 ## v0.0.3 - 2025-05-23
 - Dependencies: Updated to `sqlalchemy-cratedb==0.42.0.dev2`

pyproject.toml

@@ -80,6 +80,7 @@ dependencies = [
   "grpcio-tools<1.77",
   "protobuf<6.34",
   "pycryptodome<3.24",
+  "python-dateutil<2.10",
   "sqlalchemy-cratedb==0.42.0.dev2",
   "toolz<2",
   "zstandard<0.26",

src/cratedb_fivetran_destination/main.py

@@ -14,6 +14,7 @@
     WriteHistoryBatchProcessor,
 )
 from cratedb_fivetran_destination.model import (
+    CrateDBKnowledge,
     FieldMap,
     FivetranKnowledge,
     FivetranTable,
@@ -366,7 +367,10 @@ def _files_to_records(request, files: t.List[str]):
             for record in read_csv.decrypt_file(filename, value):
                 # Rename keys according to field map.
                 record = FieldMap.rename_keys(record)
+                # Replace magic Fivetran values.
                 FivetranKnowledge.replace_values(record)
+                # Adjust values to data types for CrateDB.
+                CrateDBKnowledge.replace_values(request, record)
                 yield record
 
     def _reflect_table(self, schema: str, table: str):

src/cratedb_fivetran_destination/model.py

@@ -1,6 +1,7 @@
 import typing as t
 from textwrap import dedent
 
+import dateutil
 import sqlalchemy as sa
 from attr import Factory
 from attrs import define
@@ -74,6 +75,7 @@ class TypeMap:
         # CrateDB can not store `DATE` types, so converge to `TIMESTAMP`.
         DataType.NAIVE_DATE: sa.TIMESTAMP(),
         DataType.NAIVE_DATETIME: sa.TIMESTAMP(),
+        DataType.NAIVE_TIME: sa.TIMESTAMP(),
         DataType.UTC_DATETIME: sa.TIMESTAMP(),
         # TODO: The parameters are coming from the API, here `input_fivetran.json`.
         #       How to loop them into this type resolution machinery?
@@ -82,7 +84,6 @@ class TypeMap:
         DataType.STRING: sa.String(),
         DataType.JSON: ObjectTypeImpl(),
         DataType.XML: sa.String(),
-        DataType.NAIVE_TIME: sa.TIMESTAMP(),
     }
 
     cratedb_map = {
@@ -109,6 +110,8 @@ class TypeMap:
         # sa.DateTime: DataType.NAIVE_DATETIME,
         sa.DateTime: DataType.UTC_DATETIME,
         sa.DATETIME: DataType.UTC_DATETIME,
+        sa.Time: DataType.NAIVE_TIME,
+        sa.TIME: DataType.NAIVE_TIME,
         sa.Numeric: DataType.DECIMAL,
         sa.DECIMAL: DataType.DECIMAL,
         sa.LargeBinary: DataType.BINARY,
@@ -156,6 +159,27 @@ def replace_values(cls, record):
             record.pop(rm)
 
 
+class CrateDBKnowledge:
+    """
+    Manage special knowledge about CrateDB.
+
+    CrateDB can't store values of the `TIME` type, so we selected to store it as `DATETIME`
+    This routine converges the value.
+    """
+
+    @classmethod
+    def replace_values(cls, request, record):
+        for column in request.table.columns:
+            if column.type == common_pb2.DataType.NAIVE_TIME and column.name in record:
+                obj = dateutil.parser.parse(record[column.name])
+                obj = obj.replace(year=1970, month=1, day=1)
+                # Calculate milliseconds since midnight (timezone-independent).
+                ms = (
+                    obj.hour * 3600 + obj.minute * 60 + obj.second
+                ) * 1000 + obj.microsecond // 1000
+                record[column.name] = str(ms)
+
+
 @define
 class TableInfo:
     """

tests/data/cratedb_canonical/input_cratedb.json

@@ -46,7 +46,7 @@
             "string": "Hotzenplotz",
             "json": "{\"foo\": \"bar\", \"count\": 42}",
             "xml": "XML",
-            "naive_time": "1970-01-02T00:00:00+01:00"
+            "naive_time": "12:34:56"
           }
         ]
       }

tests/test_integration.py

@@ -106,7 +106,7 @@ def starter():
     string="Hotzenplotz",
     json={"count": 42, "foo": "bar"},
     xml="XML",
-    naive_time=86400000,
+    naive_time=45296000,
     __fivetran_synced=mock.ANY,
     __fivetran_id="zyx-987-abc",
 )