{"record":{"id":"fce0be1a292ec1f3","repo":"apache/seatunnel","slug":"source-table-field-count-doesn-t-match-row-fi","errorCode":null,"errorMessage":"Source table field count ({}) doesn't match row field count ({}), using default column names","messagePattern":"Source table field count \\((.+?)\\) doesn't match row field count \\((.+?)\\), using default column names","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java","lineNumber":626,"sourceCode":"        } else if (typeName.contains(\"BOOLEAN\")) {\n            return BasicType.BOOLEAN_TYPE;\n        } else {\n            return BasicType.STRING_TYPE;\n        }\n    }\n\n    private SeaTunnelRowType inferRowTypeFromRow(SeaTunnelRow row) {\n        Object[] fields = row.getFields();\n        String[] fieldNames = new String[fields.length];\n        SeaTunnelDataType<?>[] fieldTypes = new SeaTunnelDataType<?>[fields.length];\n\n        // use the column names from the catalog table if available\n        if (catalogTable != null && catalogTable.getSeaTunnelRowType() != null) {\n            String[] sourceFieldNames = catalogTable.getSeaTunnelRowType().getFieldNames();\n            if (sourceFieldNames.length == fields.length) {\n                fieldNames = sourceFieldNames;\n            } else {\n                log.warn(\n                        \"Source table field count ({}) doesn't match row field count ({}), using default column names\",\n                        sourceFieldNames.length,\n                        fields.length);\n                for (int i = 0; i < fields.length; i++) {\n                    fieldNames[i] = \"column_\" + (i + 1);\n                }\n            }\n        } else {\n            // if catalog table is not available, throw an exception\n            log.warn(\"No source table schema available, can't get column names\");\n            throw new DatabendConnectorException(\n                    DatabendConnectorErrorCode.SCHEMA_NOT_FOUND,\n                    \"Source table schema is empty or null, cannot infer row type\");\n        }\n\n        for (int i = 0; i < fields.length; i++) {\n            Object field = fields[i];\n","sourceCodeStart":608,"sourceCodeEnd":644,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/sink/DatabendSinkWriter.java#L608-L644","documentation":"In DatabendSinkWriter.inferRowTypeFromRow(), the writer infers column names from the CatalogTable's row type when their field count matches the row's field count. When counts differ, it logs this warning and falls back to generic names column_1..column_N. The write proceeds, but data may land in wrong/unexpected columns since names are synthetic.","triggerScenarios":"The row produced upstream has a different number of fields than the catalog source table's schema (e.g. transform added/dropped columns, CDC rows include extra op/metadata fields, or schema evolution changed the table).","commonSituations":"Adding a transform that changes arity without updating sink schema; DDL change on the source adding a column while the job runs with the old catalog schema; CDC rows carrying extra internal fields.","solutions":["Align the pipeline schema: ensure the transform output arity matches the catalog table schema or update the sink's configured columns","Restart the job after source DDL changes so a fresh CatalogTable is captured","Enable schema evolution / schema-change support so the writer rebuilds the row type","Explicitly define the sink's target column list so inference isn't needed"],"exampleFix":"// before\nlog.warn(\"Source table field count ({}) doesn't match row field count ({}), using default column names\", sourceFieldNames.length, fields.length);\n// after\nlog.error(\"Source table field count ({}) doesn't match row field count ({}); fix upstream transform schema\", sourceFieldNames.length, fields.length);\nthrow new DatabendConnectorException(DatabendConnectorErrorCode.SCHEMA_NOT_FOUND, \"Row field count mismatch\");","handlingStrategy":"validation","validationCode":"// check pipeline arity vs source schema before submitting\nif (catalogTable.getSeaTunnelRowType().getFieldNames().length != expectedRowFields.length) {\n    throw new IllegalStateException(\"Row arity mismatch with source table schema\");\n}","typeGuard":"boolean arityMatches(CatalogTable ct, SeaTunnelRow row) {\n    return ct != null && ct.getSeaTunnelRowType() != null\n        && ct.getSeaTunnelRowType().getFieldNames().length == row.getFields().length;\n}","tryCatchPattern":"// not applicable: warning only; writer falls back to column_N names","preventionTips":["Keep transform output arity aligned with the source catalog schema","Restart jobs after source DDL changes","Explicitly configure sink column lists to avoid inference"],"tags":["schema","inference","data-mapping","arity-mismatch"],"backgroundTag":"shape-mismatch","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"}