Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Incremental sync strategy fails for timestamp/rowversion replication keys #27

Open
chrispasakarnis opened this issue Mar 30, 2020 · 1 comment
Open

Incremental sync strategy fails for timestamp/rowversion replication keys #27

chrispasakarnis opened this issue Mar 30, 2020 · 1 comment

Comments

@chrispasakarnis
Copy link

Using a timestamp / rowversion column as a replication key causes the incremental strategy to error due to the byte array that's being written to STATE being passed to the where clause.

I tested that converting the replication-key-value back into a number allows it to complete successfully.

Example state:

{"bookmarks": {"TEST_dbo_person": {"replication_key_value": [0, 0, 0, 0, 0, 0, 7, -25], "replication_key_name": "RowVers", "version": 1585608487075}}}

Test Schema:

{
    "streams": [{
        "stream": "person",
        "tap_stream_id": "TEST_dbo_person",
        "table_name": "person",
        "schema": {
            "type": "object",
            "properties": {
                "person_id": {
                    "type": ["integer"],
                    "minimum": -2147483648,
                    "maximum": 2147483647
                },
                "first_name": {
                    "type": ["string", "null"],
                    "maxLength": 50
                },
                "last_name": {
                    "type": ["string", "null"],
                    "maxLength": 50
                },
                "gender": {
                    "type": ["string", "null"],
                    "maxLength": 1
                },
                "RowVers": {
                    "type": ["string", "null"]
                },
                "Edit_DtTm": {
                    "type": ["string", "null"],
                    "format": "date-time"
                },
                "Create_DtTm": {
                    "type": ["string", "null"],
                    "format": "date-time"
                }
            }
        },
        "metadata": [{
            "metadata": {
                "database-name": "TEST",
                "schema-name": "dbo",
                "table-key-properties": ["person_id"],
                "is-view": false,
                "row-count": 1,
                "selected": true,
                "replication-method": "INCREMENTAL",
                "replication-key": "RowVers"
            },
            "breadcrumb": []
        }, {
            "metadata": {
                "inclusion": "automatic",
                "sql-datatype": "int identity",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "person_id"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "varchar",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "first_name"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "varchar",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "last_name"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "char",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "gender"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "timestamp",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "RowVers"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "datetime",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "Edit_DtTm"]
        }, {
            "metadata": {
                "inclusion": "available",
                "sql-datatype": "datetime",
                "selected-by-default": true
            },
            "breadcrumb": ["properties", "Create_DtTm"]
        }]
    }]
}

Example stack trace:

INFO Successfully connected to the instance
INFO Syncing stream TEST_dbo_person using replication method INCREMENTAL
INFO Executing query: ("SELECT [person_id], [first_name], [last_name], [gender], [RowVers], [Edit_DtTm], [Create_DtTm] FROM [dbo].[person] WHERE RowVers >= ? ORDER BY RowVers" [0 0 0 0 0 0 7 -25])
INFO Mapping: customer to ['customer']
INFO Mapping: tp_customer_person_id__sdc_sequence_idx to None
INFO Mapping: tp_person__1585608487075_person_id__sdc_sequence_idx to None
INFO Mapping: person to ['person']
INFO Stream customer (customer) with max_version None targetting None
INFO Root table name customer
INFO Writing batch with 0 records for `customer` with `key_properties`: `['person_id']`
INFO Writing table batch schema for `('customer',)`...
com.microsoft.sqlserver.jdbc.SQLServerException: The conversion from UNKNOWN to UNKNOWN is unsupported.
	at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDriverError(SQLServerException.java:228)
	at com.microsoft.sqlserver.jdbc.DataTypes.throwConversionError(DataTypes.java:1094)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.setObject(SQLServerPreparedStatement.java:1595)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.setObjectNoType(SQLServerPreparedStatement.java:1493)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.setObject(SQLServerPreparedStatement.java:1502)
	at clojure.java.jdbc$eval705$fn__706.invokePrim(jdbc.clj:484)
	at clojure.java.jdbc$eval705$fn__706.invoke(jdbc.clj)
	at clojure.java.jdbc$eval684$fn__685$G__675__694.invoke(jdbc.clj:471)
	at clojure.java.jdbc$dft_set_parameters.invokeStatic(jdbc.clj:495)
	at clojure.java.jdbc$dft_set_parameters.invoke(jdbc.clj:490)
	at clojure.java.jdbc$execute_query_with_params.invokeStatic(jdbc.clj:1071)
	at clojure.java.jdbc$execute_query_with_params.invoke(jdbc.clj:1066)
	at clojure.java.jdbc$db_query_with_resultset_STAR_.invokeStatic(jdbc.clj:1095)
	at clojure.java.jdbc$db_query_with_resultset_STAR_.invoke(jdbc.clj:1075)
	at clojure.java.jdbc$reducible_query$reify__944.reduce(jdbc.clj:1339)
	at clojure.core$reduce.invokeStatic(core.clj:6747)
	at clojure.core$reduce.invoke(core.clj:6730)
	at tap_mssql.sync_strategies.incremental$sync_and_write_messages_BANG_.invokeStatic(incremental.clj:50)
	at tap_mssql.sync_strategies.incremental$sync_and_write_messages_BANG_.invoke(incremental.clj:33)
	at tap_mssql.sync_strategies.incremental$sync_BANG_.invokeStatic(incremental.clj:66)
	at tap_mssql.sync_strategies.incremental$sync_BANG_.invoke(incremental.clj:61)
	at tap_mssql.core$dispatch_sync_by_strategy.invokeStatic(core.clj:109)
	at tap_mssql.core$dispatch_sync_by_strategy.invoke(core.clj:99)
	at tap_mssql.core$sync_stream_BANG_.invokeStatic(core.clj:123)
	at tap_mssql.core$sync_stream_BANG_.invoke(core.clj:116)
	at tap_mssql.core$maybe_sync_stream_BANG_.invokeStatic(core.clj:133)
	at tap_mssql.core$maybe_sync_stream_BANG_.invoke(core.clj:129)
	at clojure.core$partial$fn__5563.invoke(core.clj:2624)
	at clojure.core.protocols$fn__7852.invokeStatic(protocols.clj:168)
	at clojure.core.protocols$fn__7852.invoke(protocols.clj:124)
	at clojure.core.protocols$fn__7807$G__7802__7816.invoke(protocols.clj:19)
	at clojure.core.protocols$seq_reduce.invokeStatic(protocols.clj:31)
	at clojure.core.protocols$fn__7835.invokeStatic(protocols.clj:75)
	at clojure.core.protocols$fn__7835.invoke(protocols.clj:75)
	at clojure.core.protocols$fn__7781$G__7776__7794.invoke(protocols.clj:13)
	at clojure.core$reduce.invokeStatic(core.clj:6748)
	at clojure.core$reduce.invoke(core.clj:6730)
	at tap_mssql.core$do_sync.invokeStatic(core.clj:143)
	at tap_mssql.core$do_sync.invoke(core.clj:139)
	at tap_mssql.core$_main.invokeStatic(core.clj:169)
	at tap_mssql.core$_main.doInvoke(core.clj:154)
	at clojure.lang.RestFn.invoke(RestFn.java:512)
	at clojure.lang.Var.invoke(Var.java:411)
	at user$eval149.invokeStatic(form-init8490039590763633355.clj:1)
	at user$eval149.invoke(form-init8490039590763633355.clj:1)
	at clojure.lang.Compiler.eval(Compiler.java:7062)
	at clojure.lang.Compiler.eval(Compiler.java:7052)
	at clojure.lang.Compiler.load(Compiler.java:7514)
	at clojure.lang.Compiler.loadFile(Compiler.java:7452)
	at clojure.main$load_script.invokeStatic(main.clj:278)
	at clojure.main$init_opt.invokeStatic(main.clj:280)
	at clojure.main$init_opt.invoke(main.clj:280)
	at clojure.main$initialize.invokeStatic(main.clj:311)
	at clojure.main$null_opt.invokeStatic(main.clj:345)
	at clojure.main$null_opt.invoke(main.clj:342)
	at clojure.main$main.invokeStatic(main.clj:424)
	at clojure.main$main.doInvoke(main.clj:387)
	at clojure.lang.RestFn.applyTo(RestFn.java:137)
	at clojure.lang.Var.applyTo(Var.java:702)
	at clojure.main.main(main.java:37)
FATAL The conversion from UNKNOWN to UNKNOWN is unsupported.
INFO Shutting down the nrepl server
@luandy64
Copy link
Contributor

luandy64 commented Apr 9, 2020

Seems like this is related to #15 but for incremental instead of full table.

Would you like to open a PR with your fix?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

No branches or pull requests
2 participants
@luandy64 @chrispasakarnis