Handle NumberFormatException in debezium while updating sequence max …

…value for sequence on non-integer column (#2360)

Skipping the max value update for sequences that are non-Integer columns in debezium.

Fixes - https://yugabyte.atlassian.net/browse/DB-15398

debezium-server-voyager/debezium-server-voyagerexporter/src/main/java/io/debezium/server/ybexporter/SequenceObjectUpdater.java

@@ -129,8 +129,21 @@ public void processRecord(Record r){
             }
             String seqName = getFromColumnSequenceMap(schemaOrDbname, r.t.tableName, r.afterValueColumns.get(i));
             if (seqName != null){
-                Long columnValue = Long.valueOf(r.afterValueValues.get(i).toString());
-                sequenceMax.put(seqName, Math.max(sequenceMax.get(seqName), columnValue));
+                Long columnValue = null;
+                String value = r.afterValueValues.get(i).toString();
+                try {
+                    columnValue = Long.valueOf(value);
+                    sequenceMax.put(seqName, Math.max(sequenceMax.get(seqName), columnValue));
+                } catch (NumberFormatException e) {
+                    /*
+                    Skipping the sequences that are on non-Integer columns
+                    the case where Table has a text column and its default is generating a string which include the sequence nextval
+                    e.g. CREATE TABLE test(user_id text DEFAULT 'USR' || LPAD(nextval('user_code_seq')::TEXT, 4, '0'), user_name text);  
+                    For the above table the row will have USR0001 column value for user_id column 
+                    and while converting this USR0001 to Long will error out with NumberFormatException as its not a number
+                    */
+                    LOGGER.debug("Skipping unsupported sequence with non-interger value: '{}'", seqName);
+                }
             }
         }
 

migtests/tests/pg/sequences/export_data_status-report.json

@@ -93,5 +93,10 @@
       "table_name": "sales_region",
       "status": "DONE",
       "exported_count": 1000
+    },
+    {
+      "exported_count": 3,
+      "status": "DONE",
+      "table_name": "users"
     }
 ]

migtests/tests/pg/sequences/import_data_status-report.json

@@ -131,5 +131,12 @@
       "total_count": 1000,
       "imported_count": 1000,
       "percentage_complete": 100
+    },
+    {
+      "table_name": "public.\"users\"",
+      "status": "DONE",
+      "total_count": 3,
+      "imported_count": 3,
+      "percentage_complete": 100
     }
 ]

migtests/tests/pg/sequences/pg_sequences_automation.sql

@@ -214,4 +214,39 @@ WITH region_list AS (
                 amount[1 + mod(n, array_length(amount, 1))], 
                 'Branch ' || n as branch, 
                 region[1 + mod(n, array_length(region, 1))] 
-                    FROM amount_list, region_list, generate_series(1,1000) as n;
+                    FROM amount_list, region_list, generate_series(1,1000) as n;
+
+CREATE SEQUENCE user_code_seq
+START WITH 1
+INCREMENT BY 1;
+
+CREATE OR REPLACE FUNCTION generate_user_code() RETURNS TEXT AS $$
+DECLARE
+    new_code TEXT;
+BEGIN
+    SELECT 'USR' || LPAD(nextval('user_code_seq')::TEXT, 4, '0') 
+    INTO new_code;
+    RETURN new_code;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TABLE users (
+    user_id SERIAL PRIMARY KEY,
+    user_code TEXT UNIQUE,
+    name TEXT
+);
+
+ALTER TABLE ONLY users ALTER COLUMN user_code SET DEFAULT generate_user_code()
+
+ALTER SEQUENCE public.user_code_seq OWNED BY public.users.user_code;
+
+INSERT INTO users (name) 
+VALUES ('John Doe');
+
+INSERT INTO users (name) 
+VALUES ('ABC');
+
+INSERT INTO users (name) 
+VALUES ('XYZ');
+
+SELECT * FROM users;

migtests/tests/pg/sequences/validate

@@ -22,7 +22,8 @@ EXPECTED_ROW_COUNT['public'] = {
     'sales_region' : 1000,
     'boston' : 334,
     'sydney': 333,
-    'london': 333
+    'london': 333,
+    'users': 3
 }
 
 EXPECTED_ROW_COUNT['schema1'] = {
@@ -82,15 +83,15 @@ EXPECTED_TABLE_SUM['schema4'] = {
 }
 
 NUM_TABLES = {
-    'public': 13,
+    'public': 14,
     'schema1': 5,
     'schema2': 2,
     'schema3': 1,
     'schema4': 1
 }
 
 NUM_SEQUENCES = {
-    'public': 4+3+1, # 2 are implicit because of serial/generated columns
+    'public': 4+3+1+2, # 2 are implicit because of serial/generated columns
     'schema1': 1+3, # 3 are implicit because of serial/generated columns
     'schema2': 1,
     'schema3': 1,