{"record":{"id":"de673d2c2f980308","repo":"apache/hadoop","slug":"all-merged-files-must-be-compressed-or-not","errorCode":null,"errorMessage":"All merged files must be compressed or not.","messagePattern":"All merged files must be compressed or not\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java","lineNumber":3517,"sourceCode":"      private float progPerByte;\n      private Progress mergeProgress = new Progress();\n      private Path tmpDir;\n      private Progressable progress = null; //handle to the progress reporting object\n      private SegmentDescriptor minSegment;\n      \n      //a TreeMap used to store the segments sorted by size (segment offset and\n      //segment path name is used to break ties between segments of same sizes)\n      private Map<SegmentDescriptor, Void> sortedSegmentSizes =\n        new TreeMap<SegmentDescriptor, Void>();\n            \n      @SuppressWarnings(\"unchecked\")\n      public void put(SegmentDescriptor stream) throws IOException {\n        if (size() == 0) {\n          compress = stream.in.isCompressed();\n          blockCompress = stream.in.isBlockCompressed();\n        } else if (compress != stream.in.isCompressed() || \n                   blockCompress != stream.in.isBlockCompressed()) {\n          throw new IOException(\"All merged files must be compressed or not.\");\n        } \n        super.put(stream);\n      }\n      \n      /**\n       * A queue of file segments to merge\n       * @param segments the file segments to merge\n       * @param tmpDir a relative local directory to save intermediate files in\n       * @param progress the reference to the Progressable object\n       */\n      public MergeQueue(List <SegmentDescriptor> segments,\n          Path tmpDir, Progressable progress) {\n        int size = segments.size();\n        for (int i = 0; i < size; i++) {\n          sortedSegmentSizes.put(segments.get(i), null);\n        }\n        this.tmpDir = tmpDir;\n        this.progress = progress;","sourceCodeStart":3499,"sourceCodeEnd":3535,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/SequenceFile.java#L3499-L3535","documentation":"SequenceFile.Sorter's MergeQueue fixes the expected compression mode from the first segment added, then requires every subsequent segment to match both the compress flag and the blockCompress flag. Any later file that differs (compressed vs uncompressed, or RECORD vs BLOCK compression) throws IOException(\"All merged files must be compressed or not.\"). Mixing compression modes would make a single valid output header impossible.","triggerScenarios":"Calling Sorter.sort/merge/sortAndIterate over inputs where some SequenceFiles were written with CompressionType.RECORD or BLOCK and others with NONE; mixing RECORD- and BLOCK-compressed files (blockCompress flag differs even if both are 'compressed'); merging map outputs from jobs with different mapreduce.map.output.compression settings.","commonSituations":"Merging historical outputs written under older configs; inputs produced by different tools (some compress, some don't); changing io.seqfile.compression.type between runs and then sorting the combined set.","solutions":["Partition the input list by compression mode (read each header with a short-lived SequenceFile.Reader; use isCompressed()/isBlockCompressed()) and run one merge per group.","Or normalize first: rewrite every input with one chosen CompressionType and codec, then merge.","Set identical compression configuration on all producing jobs so inputs can never diverge."],"exampleFix":"// before\nsorter.merge(inFiles, outFile);           // throws when modes differ\n\n// after: group by compression mode, merge each group separately\nMap<List<Boolean>, List<Path>> groups = new HashMap<>();\nfor (Path p : inFiles) {\n  try (SequenceFile.Reader r = new SequenceFile.Reader(conf,\n      SequenceFile.Reader.file(p))) {\n    groups.computeIfAbsent(\n        Arrays.asList(r.isCompressed(), r.isBlockCompressed()),\n        k -> new ArrayList<>()).add(p);\n  }\n}\nfor (List<Path> group : groups.values()) {\n  sorter.merge(group.toArray(new Path[0]), nextUniqueOutput());\n}","handlingStrategy":"validation","validationCode":"static boolean homogeneousCompression(Configuration conf, Path[] inFiles)\n    throws IOException {\n  Boolean compress = null, block = null;\n  for (Path p : inFiles) {\n    try (SequenceFile.Reader r = new SequenceFile.Reader(conf,\n        SequenceFile.Reader.file(p))) {\n      if (compress == null) {\n        compress = r.isCompressed();\n        block = r.isBlockCompressed();\n      } else if (compress != r.isCompressed()\n                 || block != r.isBlockCompressed()) {\n        return false;\n      }\n    }\n  }\n  return true;\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Pre-check isCompressed()/isBlockCompressed() of every input header before merging","Normalize heterogeneous inputs by rewriting them with one CompressionType","Keep compression configuration identical across all producing jobs"],"tags":["sequencefile","merge","compression","incompatible-inputs","hadoop"],"backgroundTag":"incompatible-input-configuration","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-23T01:17:44.959Z"}