{"record":{"id":"ac9f3c260142e4f9","repo":"apache/seatunnel","slug":"schema-not-found","errorCode":"SCHEMA_NOT_FOUND","errorMessage":"Source table schema is empty or null","messagePattern":"Source table schema is empty or null","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/DatabendSinkWriter.java","lineNumber":197,"sourceCode":"                    DatabendConnectorErrorCode.SQL_OPERATION_FAILED,\n                    \"Raw table name not set by AggregatedCommitter. Cannot initialize CDC PreparedStatement.\");\n        }\n\n        // Generate insert SQL for raw table\n        String insertRawSql = generateInsertRawSql(sinkTablePath.getDatabaseName());\n\n        // Create the PreparedStatement\n        this.cdcPreparedStatement = connection.prepareStatement(insertRawSql);\n        this.cdcPreparedStatement.setQueryTimeout(executeTimeoutSec);\n\n        log.info(\"CDC PreparedStatement created successfully with SQL: {}\", insertRawSql);\n    }\n\n    private void initTraditionalMode(String database, String table) throws SQLException {\n        // use the catalog table schema to create the target table\n        SeaTunnelRowType rowType = catalogTable.getSeaTunnelRowType();\n        if (rowType == null || rowType.getFieldNames().length == 0) {\n            throw new DatabendConnectorException(\n                    DatabendConnectorErrorCode.SCHEMA_NOT_FOUND,\n                    \"Source table schema is empty or null\");\n        }\n\n        this.insertSql = generateInsertSql(database, table, rowType);\n        log.info(\"Generated insert SQL: {}\", insertSql);\n        try {\n            this.schemaChangeManager = new SchemaChangeManager(databendSinkConfig);\n            this.preparedStatement = connection.prepareStatement(insertSql);\n            this.preparedStatement.setQueryTimeout(executeTimeoutSec);\n            log.info(\"PreparedStatement created successfully\");\n        } catch (SQLException e) {\n            throw new DatabendConnectorException(\n                    DatabendConnectorErrorCode.SQL_OPERATION_FAILED,\n                    \"Failed to prepare statement: \" + e.getMessage(),\n                    e);\n        }\n    }","sourceCodeStart":179,"sourceCodeEnd":215,"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#L179-L215","documentation":"Thrown by initTraditionalMode when the source CatalogTable's SeaTunnelRowType is null or has zero fields. The writer needs the source schema to generate the target-table DDL and INSERT SQL, so an empty schema is a hard stop.","triggerScenarios":"catalogTable.getSeaTunnelRowType() returns null or fieldNames.length == 0 — the upstream produced a catalog table with no columns (bad source config, unsupported source type, or a transform stripped all fields).","commonSituations":"Source connector misconfigured so it emits no fields; fake/source with empty schema used in testing; a Filter/Transform projecting away all columns; API/connector version mismatch producing an unpopulated CatalogTable.","solutions":["Fix the upstream source configuration so it discovers/emits a real schema (e.g. correct table/query, metadata readable).","Check for transforms between source and sink that remove all fields.","Confirm source and sink plugin versions are compatible and the CatalogTable is populated.","Add a pre-flight check in the job: print the resolved schema via the REST API/CLI before running."],"exampleFix":"// before\nFilter transform: sql = \"SELECT 1 FROM source\" // drops all columns\n// after\nFilter transform: sql = \"SELECT id, name, ts FROM source\" // schema preserved","handlingStrategy":"validation","validationCode":"// pre-flight:\nSeaTunnelRowType rt = catalogTable.getSeaTunnelRowType();\nif (rt == null || rt.getFieldNames().length == 0) throw new IllegalArgumentException(\"Upstream produced an empty schema — fix source/transform config\");\n","typeGuard":"boolean hasSchema(CatalogTable t) { return t != null && t.getSeaTunnelRowType() != null && t.getSeaTunnelRowType().getFieldNames().length > 0; }","tryCatchPattern":null,"preventionTips":["Inspect the job plan/REST API to confirm the source resolves a non-empty schema.","Avoid transforms that project away all columns.","Test the source standalone before wiring it to the Databend sink.","Pin compatible source/transform/sink plugin versions."],"tags":["databend","schema","validation","writer"],"backgroundTag":"schema-validation-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"}