{"record":{"id":"66837942ae6121af","repo":"apache/seatunnel","slug":"sql-operation-failed-668379","errorCode":"SQL_OPERATION_FAILED","errorMessage":"Failed to execute merge operation: {e.getMessage()}","messagePattern":"Failed to execute merge operation: (.+?)","errorType":"error_code","errorClass":"DatabendConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkAggregatedCommitter.java","lineNumber":162,"sourceCode":"        return new ArrayList<>();\n    }\n\n    /** Perform merge from CDC stream to target table. */\n    private void performMerge() {\n        // Merge all the data from raw table to target table\n        String mergeSql = generateMergeSql();\n        log.info(\"[Instance {}] Executing MERGE INTO statement: {}\", instanceId, mergeSql);\n\n        try (Statement stmt = connection.createStatement()) {\n            stmt.execute(mergeSql);\n            log.info(\"[Instance {}] Merge operation completed successfully\", instanceId);\n        } catch (SQLException e) {\n            log.error(\n                    \"[Instance {}] Failed to execute merge operation: {}\",\n                    instanceId,\n                    e.getMessage(),\n                    e);\n            throw new DatabendConnectorException(\n                    DatabendConnectorErrorCode.SQL_OPERATION_FAILED,\n                    \"Failed to execute merge operation: \" + e.getMessage(),\n                    e);\n        }\n    }\n\n    private String generateMergeSql() {\n        StringBuilder sql = new StringBuilder();\n        sql.append(String.format(\"MERGE INTO %s.%s a \", database, table));\n        sql.append(\"USING (SELECT \");\n\n        // Add all columns from raw_data\n        if (catalogTable != null && catalogTable.getSeaTunnelRowType() != null) {\n            String[] fieldNames = catalogTable.getSeaTunnelRowType().getFieldNames();\n            for (int i = 0; i < fieldNames.length; i++) {\n                if (i > 0) {\n                    sql.append(\", \");\n                }","sourceCodeStart":144,"sourceCodeEnd":180,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkAggregatedCommitter.java#L144-L180","documentation":"Thrown by DatabendSinkAggregatedCommitter when the JDBC MERGE statement executed during commit (or close-time flush) fails with a SQLException. DatabendConnectorException wraps the driver message and original exception. It means the upsert into the Databend target table could not be executed.","triggerScenarios":"performMerge executes connection.prepareStatement(mergeSql).executeUpdate() and the driver throws: syntax error in generated MERGE SQL, target table missing, unique key mismatch, permission denied, connection dropped mid-commit.","commonSituations":"Merge-on-write key configured with columns that don't exist in the target table; table dropped between checkpoint and commit; network interruption to Databend during a long commit; insufficient grants for the sink user.","solutions":["Check the wrapped SQLException cause for the exact driver error and fix the MERGE SQL inputs (key columns, table names).","Verify the target table exists and the sink user has INSERT/UPDATE grants.","Test connectivity to Databend (host/port/credentials) and retry the job from the last checkpoint.","If SQL is invalid due to schema drift, recreate the target table to match the SeaTunnel catalog schema."],"exampleFix":"// before\nString mergeSql = \"MERGE INTO \" + table + \" USING source ON id = id ...\"; // ambiguous ON clause\n// after\nString mergeSql = \"MERGE INTO \" + table + \" t USING (SELECT ? AS id) s ON t.id = s.id WHEN MATCHED THEN UPDATE ...\";","handlingStrategy":"try-catch","validationCode":"// before running the job:\n// SHOW GRANTS FOR user; DESCRIBE target_table;\n// confirm merge key columns exist and user has INSERT/UPDATE privileges\n","typeGuard":null,"tryCatchPattern":"try { committer.commit(committables); } catch (DatabendConnectorException e) { if (e.getCause() instanceof SQLException sqlEx) { log.error(\"MERGE failed: {} state={}\", sqlEx.getMessage(), sqlEx.getSQLState()); } throw e; }","preventionTips":["Always wrap committer.commit() in try-catch and inspect the SQLException cause/SQLState.","Verify merge key columns exist in the target table before the job starts.","Keep checkpoints enabled so commits can be retried after transient failures.","Monitor Databend availability from workers before large commits."],"tags":["jdbc","databend","sql","commit"],"backgroundTag":"sql-query-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}