{"record":{"id":"01279c68c5acfc58","repo":"apache/flink","slug":"currently-filesystem-sink-doesn-t-support-setting","errorCode":null,"errorMessage":"Currently, filesystem sink doesn't support setting parallelism (%d) by '%s' when the input stream is not INSERT only. The row kinds of input stream are [%s]","messagePattern":"Currently, filesystem sink doesn't support setting parallelism \\((.+?)\\) by '(.+?)' when the input stream is not INSERT only\\. The row kinds of input stream are \\[(.+?)\\]","errorType":"validation","errorClass":"ValidationException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java","lineNumber":407,"sourceCode":"        if (bulkWriterFormat != null) {\n            return bulkWriterFormat.createRuntimeEncoder(\n                    sinkContext, physicalDataTypeWithoutPartitionColumns);\n        } else if (serializationFormat != null) {\n            return new SerializationSchemaAdapter(\n                    serializationFormat.createRuntimeEncoder(\n                            sinkContext, physicalDataTypeWithoutPartitionColumns));\n        } else {\n            throw new TableException(\"Can not find format factory.\");\n        }\n    }\n\n    private void checkConfiguredParallelismAllowed(ChangelogMode requestChangelogMode) {\n        final Integer parallelism = this.configuredParallelism;\n        if (parallelism == null) {\n            return;\n        }\n        if (!requestChangelogMode.containsOnly(RowKind.INSERT)) {\n            throw new ValidationException(\n                    String.format(\n                            \"Currently, filesystem sink doesn't support setting parallelism (%d) by '%s' \"\n                                    + \"when the input stream is not INSERT only. The row kinds of input stream are [%s]\",\n                            parallelism,\n                            FileSystemConnectorOptions.SINK_PARALLELISM.key(),\n                            requestChangelogMode.getContainedKinds().stream()\n                                    .map(RowKind::shortString)\n                                    .collect(Collectors.joining(\",\"))));\n        }\n    }\n\n    private static OutputFormat<RowData> createBulkWriterOutputFormat(\n            BulkWriter.Factory<RowData> factory, Path path) {\n        return new OutputFormat<RowData>() {\n\n            private static final long serialVersionUID = 1L;\n\n            private transient BulkWriter<RowData> writer;","sourceCodeStart":389,"sourceCodeEnd":425,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java#L389-L425","documentation":"FileSystemTableSink.checkConfiguredParallelismAllowed validates that if sink.parallelism is configured, the input stream must be INSERT-only (no UPDATE_BEFORE, UPDATE_AFTER, or DELETE row kinds). Filesystem sinks with explicit parallelism distribute rows across subtasks, and non-INSERT changelog rows would be split across files, breaking correctness. If the changelog contains non-INSERT kinds, a ValidationException is thrown listing the parallelism value, the option key, and the row kinds found.","triggerScenarios":"The table option sink.parallelism is set to a non-null value AND the input changelog mode contains row kinds other than INSERT. This happens when the upstream query produces an upsert or changelog stream (e.g. aggregations, joins that produce updates) that is written to a filesystem sink with explicit parallelism.","commonSituations":"Writing the output of a GROUP BY aggregation (which produces UPDATE rows) to a filesystem sink with sink.parallelism set. Upsert Kafka source feeding a filesystem sink with custom parallelism. Any changelog-producing operator upstream of a filesystem sink with parallelism configured.","solutions":["Remove the sink.parallelism option — let the filesystem sink use the input parallelism (changelog mode is then implicitly handled).","If parallelism control is essential, ensure the upstream query only produces INSERT-only rows (e.g. no aggregations or updates before the sink).","Insert a deduplication or append-only materialization step before the filesystem sink to normalize the changelog to INSERT-only."],"exampleFix":"-- before\nCREATE TABLE sink_t (a INT, cnt BIGINT) WITH (\n  'connector'='filesystem',\n  'path'='file:///out',\n  'format'='parquet',\n  'sink.parallelism'='4'\n);\nINSERT INTO sink_t SELECT a, COUNT(*) FROM source GROUP BY a;\n-- aggregation produces UPDATE rows -> ValidationException\n\n-- after\n-- remove sink.parallelism:\nCREATE TABLE sink_t (a INT, cnt BIGINT) WITH (\n  'connector'='filesystem',\n  'path'='file:///out',\n  'format'='parquet'\n);\nINSERT INTO sink_t SELECT a, COUNT(*) FROM source GROUP BY a;","handlingStrategy":"validation","validationCode":"// Before setting sink.parallelism, verify changelog mode is INSERT-only\nif (configuredParallelism != null && !changelogMode.containsOnly(RowKind.INSERT)) {\n    throw new ValidationException(\n        \"sink.parallelism is not supported for non-INSERT-only streams. \"\n        + \"Remove 'sink.parallelism' or ensure INSERT-only input.\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Do not set sink.parallelism on filesystem sinks fed by aggregation or upsert operators.","Verify the upstream query produces INSERT-only rows before configuring parallelism.","If parallelism is needed, insert a normalization step (e.g. deduplication) before the sink.","Document which pipeline patterns are incompatible with sink.parallelism."],"tags":["table-sink","parallelism","changelog-mode","validation","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}