You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@orc.apache.org by "QianyongY (via GitHub)" <gi...@apache.org> on 2026/04/15 13:29:56 UTC
[PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
QianyongY opened a new pull request, #2601:
URL: https://github.com/apache/orc/pull/2601
### What changes were proposed in this pull request?
Extends the Java merge tool so that, for inputs sharing the same schema, you can still merge to one ORC file by default, or use -m / --maxSize to write multiple ORC files under an output directory as part-xxxxx.orc, batching by on-disk input file size.
### Why are the changes needed?
As the title says: users often need to merge many compatible ORC files without producing a single huge output file. This adds an optional mode that caps output size at whole-file boundaries while keeping the existing single-file behavior when --maxSize is not set.
### How was this patch tested?
Add UT
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3128086156
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +110,348 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ long firstTwo = fs.getFileStatus(new Path(sourceNames[0])).getLen()
+ + fs.getFileStatus(new Path(sourceNames[1])).getLen();
+ long maxSize = firstTwo + 1;
+
+ Path outputDir = new Path(workDir + File.separator + "merge-multi-out");
+ fs.delete(outputDir, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{workDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Input files size: 3"), "Should report 3 input files");
+ assertTrue(output.contains("Merge files size: 3"), "All 3 files should be merged");
+ assertTrue(fs.isDirectory(outputDir), "Output directory should be created");
+
+ // Verify that multiple part files were created and total row count is correct.
+ long totalRows = 0;
+ int partCount = 0;
+ for (int i = 0; ; i++) {
+ Path part = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, i));
+ if (!fs.exists(part)) {
+ break;
+ }
+ partCount++;
+ try (Reader reader = OrcFile.createReader(part, OrcFile.readerOptions(conf))) {
+ totalRows += reader.getNumberOfRows();
+ }
+ }
+ assertEquals(2, partCount, "Expected exactly two output part files, got: " + partCount);
+ assertEquals(5000 + 5000 + 5000, totalRows, "Total row count across all parts should match");
+ }
+
+ /**
+ * A single input file that is larger than --maxSize must still be emitted as its
+ * own part file (we never split an input).
+ */
+ @Test
+ public void testMergeWithMaxSizeSingleGiantFile() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputDir = new Path(workDir, "giant-in");
+ fs.mkdirs(inputDir);
+ Path giant = new Path(inputDir, "giant.orc");
+ writeOrcFile(giant, schema, 20000);
+ long giantSize = fs.getFileStatus(giant).getLen();
+ assertTrue(giantSize > 0, "Giant source file should have non-zero size");
+
+ Path outputDir = new Path(workDir, "giant-out");
+ fs.delete(outputDir, true);
+
+ long maxSize = Math.max(1L, giantSize / 2);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
Review Comment:
These tests unconditionally print the captured tool output to stdout, which can make CI logs noisy. Consider removing this println (or only printing on assertion failure) since failures can include the captured output in the assertion message.
```suggestion
```
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -55,28 +62,76 @@ public static void main(Configuration conf, String[] args) throws Exception {
return;
Review Comment:
The message "output filename is null" is misleading here because the condition also covers the empty-string case (and more broadly this is an output *path* that can be a directory in multi-file modes). Consider changing it to something like "--output is required" / "output path is missing" to better guide CLI users.
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +110,348 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ long firstTwo = fs.getFileStatus(new Path(sourceNames[0])).getLen()
+ + fs.getFileStatus(new Path(sourceNames[1])).getLen();
+ long maxSize = firstTwo + 1;
+
+ Path outputDir = new Path(workDir + File.separator + "merge-multi-out");
+ fs.delete(outputDir, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{workDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Input files size: 3"), "Should report 3 input files");
+ assertTrue(output.contains("Merge files size: 3"), "All 3 files should be merged");
+ assertTrue(fs.isDirectory(outputDir), "Output directory should be created");
+
+ // Verify that multiple part files were created and total row count is correct.
+ long totalRows = 0;
+ int partCount = 0;
+ for (int i = 0; ; i++) {
+ Path part = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, i));
+ if (!fs.exists(part)) {
+ break;
+ }
+ partCount++;
+ try (Reader reader = OrcFile.createReader(part, OrcFile.readerOptions(conf))) {
+ totalRows += reader.getNumberOfRows();
+ }
+ }
+ assertEquals(2, partCount, "Expected exactly two output part files, got: " + partCount);
+ assertEquals(5000 + 5000 + 5000, totalRows, "Total row count across all parts should match");
+ }
+
+ /**
+ * A single input file that is larger than --maxSize must still be emitted as its
+ * own part file (we never split an input).
+ */
+ @Test
+ public void testMergeWithMaxSizeSingleGiantFile() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputDir = new Path(workDir, "giant-in");
+ fs.mkdirs(inputDir);
+ Path giant = new Path(inputDir, "giant.orc");
+ writeOrcFile(giant, schema, 20000);
+ long giantSize = fs.getFileStatus(giant).getLen();
+ assertTrue(giantSize > 0, "Giant source file should have non-zero size");
+
+ Path outputDir = new Path(workDir, "giant-out");
+ fs.delete(outputDir, true);
+
+ long maxSize = Math.max(1L, giantSize / 2);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Input files size: 1"), "Should report 1 input file");
+ assertTrue(output.contains("Merge files size: 1"), "The giant file should be merged");
+
+ Path part0 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ Path part1 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 1));
+ assertTrue(fs.exists(part0), "Expected part-00000.orc");
+ assertFalse(fs.exists(part1), "Expected exactly one part file for a single input");
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(20000, reader.getNumberOfRows(),
+ "Single giant file must be preserved intact in its own part");
+ }
+ }
+
+ /**
+ * By default running the merge against a non-empty output directory must fail so
+ * that existing data is not silently destroyed. With --overwrite the directory's
+ * contents are deleted before new part files are written.
+ */
+ @Test
+ public void testMergeWithMaxSizeOverwriteBehavior() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputDir = new Path(workDir, "ow-in");
+ fs.mkdirs(inputDir);
+ writeOrcFile(new Path(inputDir, "a.orc"), schema, 100);
+
+ Path outputDir = new Path(workDir, "ow-out");
+ fs.delete(outputDir, true);
+ fs.mkdirs(outputDir);
+ Path stale = new Path(outputDir, "stale.txt");
+ fs.create(stale).close();
+
+ String[] baseArgs = {inputDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", "1048576"};
+
+ // Without --overwrite: should reject non-empty output directory and leave it untouched.
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> MergeFiles.main(conf, baseArgs));
+ assertTrue(ex.getMessage().contains("not empty") &&
+ ex.getMessage().contains("--overwrite"),
+ "Error should mention --overwrite: " + ex.getMessage());
+ assertTrue(fs.exists(stale),
+ "Existing content must be preserved when overwrite is refused");
+
+ // With --overwrite: existing content is cleared and fresh part files are written.
+ String[] owArgs = new String[baseArgs.length + 1];
+ System.arraycopy(baseArgs, 0, owArgs, 0, baseArgs.length);
+ owArgs[baseArgs.length] = "--overwrite";
+
+ PrintStream origErr = System.err;
+ ByteArrayOutputStream myErr = new ByteArrayOutputStream();
+ System.setErr(new PrintStream(myErr, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, owArgs);
+ System.err.flush();
+ } finally {
+ System.setErr(origErr);
+ }
+ String errOut = myErr.toString(StandardCharsets.UTF_8);
+ assertTrue(errOut.contains("Overwriting existing non-empty output directory"),
+ "Expected an overwrite warning on stderr, got:\n" + errOut);
+ assertFalse(fs.exists(stale), "Pre-existing file should be cleared by --overwrite");
+
+ Path part0 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ assertTrue(fs.exists(part0), "Expected part-00000.orc after overwrite");
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(100, reader.getNumberOfRows());
+ }
+ }
+
+ /**
+ * Creates an ORC file at the given path with {@code rowCount} rows of the fixed
+ * {@code struct<x:int,y:string>} schema used across these tests.
+ */
+ private void writeOrcFile(Path path, TypeDescription schema, int rowCount) throws Exception {
+ Writer writer = OrcFile.createWriter(path,
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCount; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ /**
+ * Two-level nested partitions (d=.../h=01, h=02, h=03). With --preserveStructure,
+ * each leaf partition must be merged independently and its relative path mirrored
+ * under the output root. Data MUST NOT be cross-mixed between leaves.
+ */
+ @Test
+ public void testPreserveStructureTwoLevelPartitions() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputRoot = new Path(workDir, "ps-in");
+ Path partitionBase = new Path(inputRoot, "d=2025-04-25");
+
+ int[] hRows = {100, 200, 300};
+ for (int h = 0; h < hRows.length; h++) {
+ Path leaf = new Path(partitionBase, String.format("h=%02d", h + 1));
+ fs.mkdirs(leaf);
+ writeOrcFile(new Path(leaf, "a.orc"), schema, hRows[h]);
+ writeOrcFile(new Path(leaf, "b.orc"), schema, hRows[h] / 2);
+ }
+
+ Path outputRoot = new Path(workDir, "ps-out");
+ fs.delete(outputRoot, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputRoot.toString(),
+ "--output", outputRoot.toString(),
+ "--preserveStructure"});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
Review Comment:
These tests unconditionally print the captured tool output to stdout, which can make test runs noisy. Consider removing this println (or only printing on failure) and rely on assertion messages to surface the output when needed.
```suggestion
```
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -55,28 +62,76 @@ public static void main(Configuration conf, String[] args) throws Exception {
return;
}
boolean ignoreExtension = cli.hasOption("ignoreExtension");
+ boolean preserveStructure = cli.hasOption("preserveStructure");
+ boolean overwrite = cli.hasOption("overwrite");
- List<Path> inputFiles = new ArrayList<>();
- OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf);
+ long maxSizeBytes = 0;
+ if (cli.hasOption("maxSize")) {
+ try {
+ maxSizeBytes = Long.parseLong(cli.getOptionValue("maxSize"));
+ } catch (NumberFormatException e) {
+ System.err.println("--maxSize requires a numeric value in bytes.");
+ System.exit(1);
+ }
+ if (maxSizeBytes <= 0) {
+ System.err.println("--maxSize must be a positive number of bytes.");
+ System.exit(1);
+ }
+ }
String[] files = cli.getArgs();
+ OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf);
+
+ if (preserveStructure) {
+ if (files.length != 1) {
+ System.err.println(
+ "--preserveStructure requires exactly one input directory.");
+ System.exit(1);
+ }
+ mergePreserveStructure(conf, writerOptions, new Path(files[0]),
+ new Path(outputFilename), maxSizeBytes, ignoreExtension, overwrite);
+ return;
+ }
+
+ List<LocatedFileStatus> inputStatuses = new ArrayList<>();
for (String root : files) {
Path rootPath = new Path(root);
FileSystem fs = rootPath.getFileSystem(conf);
for (RemoteIterator<LocatedFileStatus> itr = fs.listFiles(rootPath, true); itr.hasNext(); ) {
LocatedFileStatus status = itr.next();
if (status.isFile() && (ignoreExtension || status.getPath().getName().endsWith(".orc"))) {
- inputFiles.add(status.getPath());
+ inputStatuses.add(status);
}
}
}
- if (inputFiles.isEmpty()) {
+ if (inputStatuses.isEmpty()) {
System.err.println("No files found.");
System.exit(1);
}
- List<Path> mergedFiles = OrcFile.mergeFiles(
- new Path(outputFilename), writerOptions, inputFiles);
+ inputStatuses.sort(Comparator.comparing(FileStatus::getPath));
+
+ List<Path> inputFiles = new ArrayList<>(inputStatuses.size());
+ for (LocatedFileStatus s : inputStatuses) {
+ inputFiles.add(s.getPath());
+ }
+
+ if (maxSizeBytes > 0) {
+ mergeIntoMultipleFiles(conf, writerOptions, inputStatuses, inputFiles,
+ new Path(outputFilename), maxSizeBytes, overwrite);
+ } else {
+ mergeIntoSingleFile(writerOptions, inputFiles, new Path(outputFilename), outputFilename);
Review Comment:
With --maxSize, inputs are enumerated before the output directory is validated/cleaned (prepareOutputDir happens later). If the output directory is inside an input root and --overwrite is used, you can end up deleting files that were already included in the input list (including stale prior outputs), causing missing inputs or incorrect merges. Consider validating/cleaning the output directory before scanning inputs and/or explicitly excluding the output directory subtree from enumeration.
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +155,346 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes,
Review Comment:
mergeIntoMultipleFiles takes both inputStatuses and inputFiles, but inputFiles is just a projection of inputStatuses and is only used for inputFiles.size() in the summary. This duplication makes the API easier to misuse (mismatched lists). Consider dropping the inputFiles parameter and using inputStatuses.size() (or deriving paths locally) to keep the method contract consistent.
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +110,348 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ long firstTwo = fs.getFileStatus(new Path(sourceNames[0])).getLen()
+ + fs.getFileStatus(new Path(sourceNames[1])).getLen();
+ long maxSize = firstTwo + 1;
+
+ Path outputDir = new Path(workDir + File.separator + "merge-multi-out");
+ fs.delete(outputDir, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{workDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
Review Comment:
These tests unconditionally print the captured tool output to stdout, which can make CI logs noisy and harder to read. Consider removing this println (or only printing on assertion failure) since the assertions already include helpful context messages.
```suggestion
```
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +110,348 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ long firstTwo = fs.getFileStatus(new Path(sourceNames[0])).getLen()
+ + fs.getFileStatus(new Path(sourceNames[1])).getLen();
+ long maxSize = firstTwo + 1;
+
+ Path outputDir = new Path(workDir + File.separator + "merge-multi-out");
+ fs.delete(outputDir, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{workDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Input files size: 3"), "Should report 3 input files");
+ assertTrue(output.contains("Merge files size: 3"), "All 3 files should be merged");
+ assertTrue(fs.isDirectory(outputDir), "Output directory should be created");
+
+ // Verify that multiple part files were created and total row count is correct.
+ long totalRows = 0;
+ int partCount = 0;
+ for (int i = 0; ; i++) {
+ Path part = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, i));
+ if (!fs.exists(part)) {
+ break;
+ }
+ partCount++;
+ try (Reader reader = OrcFile.createReader(part, OrcFile.readerOptions(conf))) {
+ totalRows += reader.getNumberOfRows();
+ }
+ }
+ assertEquals(2, partCount, "Expected exactly two output part files, got: " + partCount);
+ assertEquals(5000 + 5000 + 5000, totalRows, "Total row count across all parts should match");
+ }
+
+ /**
+ * A single input file that is larger than --maxSize must still be emitted as its
+ * own part file (we never split an input).
+ */
+ @Test
+ public void testMergeWithMaxSizeSingleGiantFile() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputDir = new Path(workDir, "giant-in");
+ fs.mkdirs(inputDir);
+ Path giant = new Path(inputDir, "giant.orc");
+ writeOrcFile(giant, schema, 20000);
+ long giantSize = fs.getFileStatus(giant).getLen();
+ assertTrue(giantSize > 0, "Giant source file should have non-zero size");
+
+ Path outputDir = new Path(workDir, "giant-out");
+ fs.delete(outputDir, true);
+
+ long maxSize = Math.max(1L, giantSize / 2);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Input files size: 1"), "Should report 1 input file");
+ assertTrue(output.contains("Merge files size: 1"), "The giant file should be merged");
+
+ Path part0 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ Path part1 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 1));
+ assertTrue(fs.exists(part0), "Expected part-00000.orc");
+ assertFalse(fs.exists(part1), "Expected exactly one part file for a single input");
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(20000, reader.getNumberOfRows(),
+ "Single giant file must be preserved intact in its own part");
+ }
+ }
+
+ /**
+ * By default running the merge against a non-empty output directory must fail so
+ * that existing data is not silently destroyed. With --overwrite the directory's
+ * contents are deleted before new part files are written.
+ */
+ @Test
+ public void testMergeWithMaxSizeOverwriteBehavior() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputDir = new Path(workDir, "ow-in");
+ fs.mkdirs(inputDir);
+ writeOrcFile(new Path(inputDir, "a.orc"), schema, 100);
+
+ Path outputDir = new Path(workDir, "ow-out");
+ fs.delete(outputDir, true);
+ fs.mkdirs(outputDir);
+ Path stale = new Path(outputDir, "stale.txt");
+ fs.create(stale).close();
+
+ String[] baseArgs = {inputDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", "1048576"};
+
+ // Without --overwrite: should reject non-empty output directory and leave it untouched.
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> MergeFiles.main(conf, baseArgs));
+ assertTrue(ex.getMessage().contains("not empty") &&
+ ex.getMessage().contains("--overwrite"),
+ "Error should mention --overwrite: " + ex.getMessage());
+ assertTrue(fs.exists(stale),
+ "Existing content must be preserved when overwrite is refused");
+
+ // With --overwrite: existing content is cleared and fresh part files are written.
+ String[] owArgs = new String[baseArgs.length + 1];
+ System.arraycopy(baseArgs, 0, owArgs, 0, baseArgs.length);
+ owArgs[baseArgs.length] = "--overwrite";
+
+ PrintStream origErr = System.err;
+ ByteArrayOutputStream myErr = new ByteArrayOutputStream();
+ System.setErr(new PrintStream(myErr, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, owArgs);
+ System.err.flush();
+ } finally {
+ System.setErr(origErr);
+ }
+ String errOut = myErr.toString(StandardCharsets.UTF_8);
+ assertTrue(errOut.contains("Overwriting existing non-empty output directory"),
+ "Expected an overwrite warning on stderr, got:\n" + errOut);
+ assertFalse(fs.exists(stale), "Pre-existing file should be cleared by --overwrite");
+
+ Path part0 = new Path(outputDir, String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ assertTrue(fs.exists(part0), "Expected part-00000.orc after overwrite");
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(100, reader.getNumberOfRows());
+ }
+ }
+
+ /**
+ * Creates an ORC file at the given path with {@code rowCount} rows of the fixed
+ * {@code struct<x:int,y:string>} schema used across these tests.
+ */
+ private void writeOrcFile(Path path, TypeDescription schema, int rowCount) throws Exception {
+ Writer writer = OrcFile.createWriter(path,
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCount; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ /**
+ * Two-level nested partitions (d=.../h=01, h=02, h=03). With --preserveStructure,
+ * each leaf partition must be merged independently and its relative path mirrored
+ * under the output root. Data MUST NOT be cross-mixed between leaves.
+ */
+ @Test
+ public void testPreserveStructureTwoLevelPartitions() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputRoot = new Path(workDir, "ps-in");
+ Path partitionBase = new Path(inputRoot, "d=2025-04-25");
+
+ int[] hRows = {100, 200, 300};
+ for (int h = 0; h < hRows.length; h++) {
+ Path leaf = new Path(partitionBase, String.format("h=%02d", h + 1));
+ fs.mkdirs(leaf);
+ writeOrcFile(new Path(leaf, "a.orc"), schema, hRows[h]);
+ writeOrcFile(new Path(leaf, "b.orc"), schema, hRows[h] / 2);
+ }
+
+ Path outputRoot = new Path(workDir, "ps-out");
+ fs.delete(outputRoot, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputRoot.toString(),
+ "--output", outputRoot.toString(),
+ "--preserveStructure"});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Leaves: 3"),
+ "Expected 3 leaves in summary, got:\n" + output);
+
+ for (int h = 0; h < hRows.length; h++) {
+ Path outLeaf = new Path(outputRoot, "d=2025-04-25/" + String.format("h=%02d", h + 1));
+ assertTrue(fs.isDirectory(outLeaf),
+ "Expected mirrored leaf directory to exist: " + outLeaf);
+ Path part0 = new Path(outLeaf, String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ assertTrue(fs.exists(part0),
+ "Expected part-00000.orc under leaf: " + outLeaf);
+ long expectedRows = hRows[h] + hRows[h] / 2;
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(expectedRows, reader.getNumberOfRows(),
+ "Row count mismatch for " + outLeaf);
+ assertEquals(schema, reader.getSchema());
+ }
+ }
+ }
+
+ /**
+ * A directory containing BOTH ORC files and subdirectories is ambiguous under
+ * --preserveStructure and must be rejected before any output is written.
+ */
+ @Test
+ public void testPreserveStructureRejectsMixedDirectory() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputRoot = new Path(workDir, "psmix-in");
+ fs.mkdirs(inputRoot);
+ writeOrcFile(new Path(inputRoot, "stray.orc"), schema, 10);
+ Path subLeaf = new Path(inputRoot, "h=01");
+ fs.mkdirs(subLeaf);
+ writeOrcFile(new Path(subLeaf, "a.orc"), schema, 20);
+
+ Path outputRoot = new Path(workDir, "psmix-out");
+ fs.delete(outputRoot, true);
+
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> MergeFiles.main(conf, new String[]{inputRoot.toString(),
+ "--output", outputRoot.toString(),
+ "--preserveStructure"}));
+ assertTrue(ex.getMessage().contains("both ORC files and subdirectories"),
+ "Unexpected error message: " + ex.getMessage());
+ }
+
+ /**
+ * Hidden files/directories (names starting with '_' or '.') must always be
+ * ignored, so files like _SUCCESS or _temporary/*.orc don't pollute the merged
+ * output.
+ */
+ @Test
+ public void testPreserveStructureSkipsHidden() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ Path inputRoot = new Path(workDir, "pshidden-in");
+ Path leaf = new Path(inputRoot, "h=01");
+ fs.mkdirs(leaf);
+ writeOrcFile(new Path(leaf, "data.orc"), schema, 30);
+ // _SUCCESS-style hidden file next to the real data — should be ignored by default.
+ writeOrcFile(new Path(leaf, "_hidden.orc"), schema, 999);
+ // Hidden sibling directory (e.g. _temporary) whose contents would otherwise
+ // appear as a second leaf — should be skipped entirely.
+ Path hiddenDir = new Path(inputRoot, "_temporary");
+ fs.mkdirs(hiddenDir);
+ writeOrcFile(new Path(hiddenDir, "a.orc"), schema, 7);
+
+ Path outputRoot = new Path(workDir, "pshidden-out");
+ fs.delete(outputRoot, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ try {
+ MergeFiles.main(conf, new String[]{inputRoot.toString(),
+ "--output", outputRoot.toString(),
+ "--preserveStructure"});
+ System.out.flush();
+ } finally {
+ System.setOut(origOut);
+ }
+ String output = myOut.toString(StandardCharsets.UTF_8);
+ System.out.println(output);
+
+ assertTrue(output.contains("Leaves: 1"),
+ "Expected exactly 1 leaf (hidden dir ignored), got:\n" + output);
+ assertFalse(fs.exists(new Path(outputRoot, "_temporary")),
+ "Hidden input directory should not be mirrored to output");
+
+ Path part0 = new Path(outputRoot, "h=01/" + String.format(MergeFiles.PART_FILE_FORMAT, 0));
+ assertTrue(fs.exists(part0));
+ try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
+ assertEquals(30, reader.getNumberOfRows(),
+ "Hidden _hidden.orc rows must not be merged in");
Review Comment:
These tests unconditionally print the captured tool output to stdout, which can add noise to CI logs. Consider removing this println (or printing only when a test fails) and include output in assertion failure messages instead.
```suggestion
assertTrue(output.contains("Leaves: 1"),
"Expected exactly 1 leaf (hidden dir ignored), got:\n" + output);
assertFalse(fs.exists(new Path(outputRoot, "_temporary")),
"Hidden input directory should not be mirrored to output.\nTool output:\n" + output);
Path part0 = new Path(outputRoot, "h=01/" + String.format(MergeFiles.PART_FILE_FORMAT, 0));
assertTrue(fs.exists(part0),
"Expected merged output file to exist: " + part0 + "\nTool output:\n" + output);
try (Reader reader = OrcFile.createReader(part0, OrcFile.readerOptions(conf))) {
assertEquals(30, reader.getNumberOfRows(),
"Hidden _hidden.orc rows must not be merged in.\nTool output:\n" + output);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119756981
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +137,88 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ if (outFs.exists(outputDir)) {
+ if (!outFs.getFileStatus(outputDir).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Output path already exists and is not a directory: " + outputDir);
+ }
+ if (outFs.listStatus(outputDir).length > 0) {
+ throw new IllegalArgumentException(
+ "Output directory must be empty for multi-file merge: " + outputDir);
+ }
+ } else if (!outFs.mkdirs(outputDir)) {
+ throw new IllegalStateException("Failed to create output directory: " + outputDir);
+ }
+
+ // Group input files into batches where each batch's total size <= maxSizeBytes.
+ List<List<Path>> batches = new ArrayList<>();
+ List<Path> currentBatch = new ArrayList<>();
+ long currentBatchSize = 0;
+
+ for (LocatedFileStatus status : inputStatuses) {
+ long fileSize = status.getLen();
+ if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
Review Comment:
Actually, I understand this is practical. However, can we avoid the chance of overflow, @QianyongY ?
```
- if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
+ if (!currentBatch.isEmpty() && currentBatchSize > maxSizeBytes - fileSize) {
```
Apache ORC community had a bad experience about overflow like the following.
- https://github.com/apache/orc/commit/589d5cab8a27aac94aa4aa2ebc50504d4875b258
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3117482616
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +137,88 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ if (outFs.exists(outputDir)) {
+ if (!outFs.getFileStatus(outputDir).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Output path already exists and is not a directory: " + outputDir);
+ }
+ if (outFs.listStatus(outputDir).length > 0) {
+ throw new IllegalArgumentException(
+ "Output directory must be empty for multi-file merge: " + outputDir);
+ }
+ } else if (!outFs.mkdirs(outputDir)) {
+ throw new IllegalStateException("Failed to create output directory: " + outputDir);
+ }
+
+ // Group input files into batches where each batch's total size <= maxSizeBytes.
+ List<List<Path>> batches = new ArrayList<>();
+ List<Path> currentBatch = new ArrayList<>();
+ long currentBatchSize = 0;
+
+ for (LocatedFileStatus status : inputStatuses) {
+ long fileSize = status.getLen();
+ if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
+ batches.add(currentBatch);
+ currentBatch = new ArrayList<>();
+ currentBatchSize = 0;
+ }
+ currentBatch.add(status.getPath());
+ currentBatchSize += fileSize;
+ }
+ if (!currentBatch.isEmpty()) {
+ batches.add(currentBatch);
+ }
+
+ int totalMerged = 0;
+ List<Path> allUnmerged = new ArrayList<>();
+
+ for (int i = 0; i < batches.size(); i++) {
+ List<Path> batch = batches.get(i);
+ Path partOutput = new Path(outputDir, String.format(PART_FILE_FORMAT, i));
+ List<Path> merged = OrcFile.mergeFiles(partOutput, writerOptions.clone(), batch);
+ totalMerged += merged.size();
+
+ if (merged.size() != batch.size()) {
+ Set<Path> mergedSet = new HashSet<>(merged);
+ for (Path p : batch) {
+ if (!mergedSet.contains(p)) {
+ allUnmerged.add(p);
+ }
+ }
+ }
+ }
+
Review Comment:
In multi-output mode, each batch is merged independently and the first file in each batch becomes the schema/reference for that part. That means a mixed-schema input set could be split into batches such that every batch is internally compatible, leading to a successful run that produces part files with different schemas—contradicting the tool/docs claim that inputs share the same schema. Consider validating that all input files share the same schema (at minimum) before batching, and failing (or listing as unmergeable) any files that differ, so behavior matches single-output mode expectations.
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +137,88 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ if (outFs.exists(outputDir)) {
+ if (!outFs.getFileStatus(outputDir).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Output path already exists and is not a directory: " + outputDir);
+ }
+ if (outFs.listStatus(outputDir).length > 0) {
+ throw new IllegalArgumentException(
+ "Output directory must be empty for multi-file merge: " + outputDir);
+ }
+ } else if (!outFs.mkdirs(outputDir)) {
+ throw new IllegalStateException("Failed to create output directory: " + outputDir);
+ }
Review Comment:
mergeIntoMultipleFiles throws IllegalArgumentException/IllegalStateException for common CLI misuse (output path exists and is not a directory, output dir non-empty, mkdirs failure). Since MergeFiles is a CLI tool, these will surface as stack traces rather than a clean one-line error and exit code. Consider catching these conditions in main and printing a user-friendly message (and optionally help) before exiting non-zero, consistent with the rest of the tool’s error handling.
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -56,27 +60,60 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
boolean ignoreExtension = cli.hasOption("ignoreExtension");
- List<Path> inputFiles = new ArrayList<>();
- OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf);
+ long maxSizeBytes = 0;
+ if (cli.hasOption("maxSize")) {
+ try {
+ maxSizeBytes = Long.parseLong(cli.getOptionValue("maxSize"));
+ } catch (NumberFormatException e) {
+ System.err.println("--maxSize requires a numeric value in bytes.");
+ System.exit(1);
+ }
+ if (maxSizeBytes <= 0) {
+ System.err.println("--maxSize must be a positive number of bytes.");
+ System.exit(1);
+ }
+ }
+ List<LocatedFileStatus> inputStatuses = new ArrayList<>();
String[] files = cli.getArgs();
for (String root : files) {
Path rootPath = new Path(root);
FileSystem fs = rootPath.getFileSystem(conf);
for (RemoteIterator<LocatedFileStatus> itr = fs.listFiles(rootPath, true); itr.hasNext(); ) {
LocatedFileStatus status = itr.next();
if (status.isFile() && (ignoreExtension || status.getPath().getName().endsWith(".orc"))) {
- inputFiles.add(status.getPath());
+ inputStatuses.add(status);
}
}
}
Review Comment:
Batching depends on the iteration order returned by FileSystem#listFiles, which is not guaranteed to be stable. With --maxSize this can change which inputs land in which part file across runs/filesystems, making outputs non-deterministic. Consider sorting inputStatuses (e.g., by path, or by path then length) before creating batches to make part grouping reproducible.
##########
site/_docs/java-tools.md:
##########
@@ -356,13 +356,39 @@ ______________________________________________________________________
## Java Merge
-The merge command can merge multiple ORC files that all have the same schema into a single ORC file.
+The merge command can merge multiple ORC files that all have the same schema. By default
+it writes a single output file. If `--maxSize` is set, `--output` is treated as a directory
+and the tool writes multiple part files (`part-00000.orc`, `part-00001.orc`, …) under it.
+Input files are grouped using their on-disk sizes so that each part’s total input size
+does not exceed the given threshold (a single input file larger than the threshold is still
+merged into its own part).
+
+`-h,--help`
+ : Print help
+
+`-i,--ignoreExtension`
+ : Include files that do not end in `.orc`
+
+`-m,--maxSize <bytes>`
+ : Maximum size in bytes for each output part; enables multi-file output under `--output`
+
+`-o,--output <path>`
+ : Output ORC filename (single-file mode) or output directory (when `--maxSize` is set)
+
Review Comment:
The implementation requires the multi-output `--output` directory to already be empty (it errors if it exists and has any contents). The docs describe `--output` as a directory when `--maxSize` is set, but don’t mention the emptiness requirement, which will surprise users. Please document this constraint (or relax it in code by writing into a subdirectory / allowing overwrite semantics).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119721091
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -113,6 +227,16 @@ private static Options createOptions() {
.desc("Ignore ORC file extension")
.build());
+ result.addOption(Option.builder("m")
+ .longOpt("maxSize")
Review Comment:
What happens when `maxSize` is `0` or negative?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119703792
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -56,27 +60,60 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
boolean ignoreExtension = cli.hasOption("ignoreExtension");
- List<Path> inputFiles = new ArrayList<>();
- OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf);
+ long maxSizeBytes = 0;
+ if (cli.hasOption("maxSize")) {
+ try {
+ maxSizeBytes = Long.parseLong(cli.getOptionValue("maxSize"));
+ } catch (NumberFormatException e) {
+ System.err.println("--maxSize requires a numeric value in bytes.");
+ System.exit(1);
+ }
+ if (maxSizeBytes <= 0) {
+ System.err.println("--maxSize must be a positive number of bytes.");
+ System.exit(1);
+ }
+ }
+ List<LocatedFileStatus> inputStatuses = new ArrayList<>();
String[] files = cli.getArgs();
for (String root : files) {
Path rootPath = new Path(root);
FileSystem fs = rootPath.getFileSystem(conf);
for (RemoteIterator<LocatedFileStatus> itr = fs.listFiles(rootPath, true); itr.hasNext(); ) {
LocatedFileStatus status = itr.next();
if (status.isFile() && (ignoreExtension || status.getPath().getName().endsWith(".orc"))) {
- inputFiles.add(status.getPath());
+ inputStatuses.add(status);
}
}
}
- if (inputFiles.isEmpty()) {
+ if (inputStatuses.isEmpty()) {
System.err.println("No files found.");
System.exit(1);
}
- List<Path> mergedFiles = OrcFile.mergeFiles(
- new Path(outputFilename), writerOptions, inputFiles);
+ List<Path> inputFiles = new ArrayList<>(inputStatuses.size());
+ for (LocatedFileStatus s : inputStatuses) {
+ inputFiles.add(s.getPath());
+ }
Review Comment:
Shall we sort `inputFiles` here by path to be more deterministic?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #2601: ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files
URL: https://github.com/apache/orc/pull/2601
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "QianyongY (via GitHub)" <gi...@apache.org>.
QianyongY commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4295054831
Thanks @cxzl25 and @dongjoon-hyun . Latest push:
- New `--preserveStructure` mode (partition-tree-aware merge, skips hidden entries).
- Deterministic ordering, overflow-safe batching, `--maxSize` docs clarified.
- Non-empty output dirs now abort by default; opt in with `--overwrite` to
delete existing contents. Shared between `--maxSize` and `--preserveStructure`.
- Tests added for all of the above. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119756981
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +137,88 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ if (outFs.exists(outputDir)) {
+ if (!outFs.getFileStatus(outputDir).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Output path already exists and is not a directory: " + outputDir);
+ }
+ if (outFs.listStatus(outputDir).length > 0) {
+ throw new IllegalArgumentException(
+ "Output directory must be empty for multi-file merge: " + outputDir);
+ }
+ } else if (!outFs.mkdirs(outputDir)) {
+ throw new IllegalStateException("Failed to create output directory: " + outputDir);
+ }
+
+ // Group input files into batches where each batch's total size <= maxSizeBytes.
+ List<List<Path>> batches = new ArrayList<>();
+ List<Path> currentBatch = new ArrayList<>();
+ long currentBatchSize = 0;
+
+ for (LocatedFileStatus status : inputStatuses) {
+ long fileSize = status.getLen();
+ if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
Review Comment:
Actually, I understand this is practical. However, can we avoid the chance of overflow, @QianyongY ?
```java
- if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
+ if (!currentBatch.isEmpty() && currentBatchSize > maxSizeBytes - fileSize) {
```
Apache ORC community had a bad experience about overflow like the following.
- https://github.com/apache/orc/commit/589d5cab8a27aac94aa4aa2ebc50504d4875b258
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4453501248
Merged to main/2.3
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119734536
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +137,88 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ if (outFs.exists(outputDir)) {
+ if (!outFs.getFileStatus(outputDir).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Output path already exists and is not a directory: " + outputDir);
+ }
+ if (outFs.listStatus(outputDir).length > 0) {
+ throw new IllegalArgumentException(
+ "Output directory must be empty for multi-file merge: " + outputDir);
Review Comment:
In this case, shall we behave as an **overwrite** for a better UX? I guess you can delete the directory recursively.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4453507273
Thank you, @QianyongY and @cxzl25 .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "cxzl25 (via GitHub)" <gi...@apache.org>.
cxzl25 commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3117582102
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +108,85 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ // Measure the size of the first source file to compute a threshold that forces a split.
+ long singleFileSize = fs.getFileStatus(new Path(sourceNames[0])).getLen();
+ // Threshold: slightly larger than one file so at most one file fits per part.
+ long maxSize = singleFileSize + 1;
Review Comment:
Is it possible to group by the first two file sizes + 1 to test if the merge really works？
sourceNames[0] len + sourceNames[1] len +1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3086947980
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +132,77 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ outFs.mkdirs(outputDir);
+
+ // Group input files into batches where each batch's total size <= maxSizeBytes.
+ List<List<Path>> batches = new ArrayList<>();
+ List<Path> currentBatch = new ArrayList<>();
+ long currentBatchSize = 0;
+
+ for (LocatedFileStatus status : inputStatuses) {
+ long fileSize = status.getLen();
+ if (!currentBatch.isEmpty() && currentBatchSize + fileSize > maxSizeBytes) {
+ batches.add(currentBatch);
+ currentBatch = new ArrayList<>();
+ currentBatchSize = 0;
+ }
+ currentBatch.add(status.getPath());
+ currentBatchSize += fileSize;
+ }
+ if (!currentBatch.isEmpty()) {
+ batches.add(currentBatch);
+ }
+
+ int totalMerged = 0;
+ List<Path> allUnmerged = new ArrayList<>();
+
+ for (int i = 0; i < batches.size(); i++) {
+ List<Path> batch = batches.get(i);
+ Path partOutput = new Path(outputDir, String.format(PART_FILE_FORMAT, i));
+ List<Path> merged = OrcFile.mergeFiles(partOutput, OrcFile.writerOptions(conf), batch);
Review Comment:
`mergeIntoMultipleFiles` takes `writerOptions` but ignores it and instead creates new `OrcFile.writerOptions(conf)` for each batch. This is confusing and can also drop caller-specified options (eg overwrite/key provider) if they are added later. Consider either removing the unused parameter, or (preferably) cloning the passed `writerOptions` for each batch (WriterOptions supports `clone()`) and passing the clone into `OrcFile.mergeFiles`.
```suggestion
OrcFile.WriterOptions batchWriterOptions = writerOptions.clone();
List<Path> merged = OrcFile.mergeFiles(partOutput, batchWriterOptions, batch);
```
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +108,82 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files with different row counts.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
+ Writer writer = OrcFile.createWriter(new Path(sourceNames[f]),
+ OrcFile.writerOptions(conf).setSchema(schema));
+ VectorizedRowBatch batch = schema.createRowBatch();
+ LongColumnVector x = (LongColumnVector) batch.cols[0];
+ BytesColumnVector y = (BytesColumnVector) batch.cols[1];
+ for (int r = 0; r < rowCounts[f]; ++r) {
+ int row = batch.size++;
+ x.vector[row] = r;
+ byte[] buffer = ("val-" + r).getBytes();
+ y.setRef(row, buffer, 0, buffer.length);
+ if (batch.size == batch.getMaxSize()) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ }
+ writer.close();
+ }
+
+ // Measure the size of the first source file to compute a threshold that forces a split.
+ long singleFileSize = fs.getFileStatus(new Path(sourceNames[0])).getLen();
+ // Threshold: slightly larger than one file so at most one file fits per part.
+ long maxSize = singleFileSize + 1;
+
+ Path outputDir = new Path(workDir + File.separator + "merge-multi-out");
+ fs.delete(outputDir, true);
+
+ PrintStream origOut = System.out;
+ ByteArrayOutputStream myOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(myOut, false, StandardCharsets.UTF_8));
+ MergeFiles.main(conf, new String[]{workDir.toString(),
+ "--output", outputDir.toString(),
+ "--maxSize", String.valueOf(maxSize)});
+ System.out.flush();
+ System.setOut(origOut);
Review Comment:
`System.setOut(...)` is not restored in a `finally` block. If `MergeFiles.main` throws, stdout will remain redirected and may break subsequent tests. Wrap the stdout redirection/restoration in `try/finally` (or use a helper/extension) so it is always restored.
```suggestion
try {
MergeFiles.main(conf, new String[]{workDir.toString(),
"--output", outputDir.toString(),
"--maxSize", String.valueOf(maxSize)});
System.out.flush();
} finally {
System.setOut(origOut);
}
```
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +132,77 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ outFs.mkdirs(outputDir);
Review Comment:
In multi-file mode the code calls `outFs.mkdirs(outputDir)` without validating what already exists at `outputDir`. If `--output` points to an existing file (or an existing non-empty directory), later writes may fail in confusing ways or partially succeed. Consider checking for `outFs.exists(outputDir)` / `outFs.isFile(outputDir)` and either failing fast with a clear message or requiring an empty directory.
```suggestion
if (outFs.exists(outputDir)) {
if (!outFs.getFileStatus(outputDir).isDirectory()) {
throw new IllegalArgumentException(
"Output path already exists and is not a directory: " + outputDir);
}
if (outFs.listStatus(outputDir).length > 0) {
throw new IllegalArgumentException(
"Output directory must be empty for multi-file merge: " + outputDir);
}
} else if (!outFs.mkdirs(outputDir)) {
throw new IllegalStateException("Failed to create output directory: " + outputDir);
}
```
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -56,27 +60,55 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
boolean ignoreExtension = cli.hasOption("ignoreExtension");
- List<Path> inputFiles = new ArrayList<>();
- OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf);
+ long maxSizeBytes = 0;
+ if (cli.hasOption("maxSize")) {
+ maxSizeBytes = Long.parseLong(cli.getOptionValue("maxSize"));
+ if (maxSizeBytes <= 0) {
+ System.err.println("--maxSize must be a positive number of bytes.");
+ System.exit(1);
+ }
+ }
Review Comment:
`Long.parseLong(cli.getOptionValue("maxSize"))` will throw `NumberFormatException` for non-numeric inputs and currently results in an uncaught exception/stack trace. Consider catching the parse error, printing a clear message (and help text if appropriate), and exiting with a non-zero status.
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +108,82 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
+ TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+
+ // Create 3 source ORC files with different row counts.
+ String[] sourceNames = {
+ workDir + File.separator + "ms-1.orc",
+ workDir + File.separator + "ms-2.orc",
+ workDir + File.separator + "ms-3.orc"
+ };
+ int[] rowCounts = {5000, 5000, 5000};
+ for (int f = 0; f < sourceNames.length; f++) {
Review Comment:
The comment says the 3 source ORC files have "different row counts", but `rowCounts` is `{5000, 5000, 5000}`. Either update the comment or vary the row counts so the test matches its description.
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -113,6 +211,15 @@ private static Options createOptions() {
.desc("Ignore ORC file extension")
.build());
+ result.addOption(Option.builder("m")
+ .longOpt("maxSize")
+ .desc("Maximum size in bytes for each output ORC file. When set, --output is treated as "
+ + "an output directory and merged files are written as part-00000.orc, "
+ + "part-00001.orc, etc. Files are grouped at file boundaries so an individual "
+ + "file larger than this threshold will still be placed in its own part.")
+ .hasArg()
+ .build());
Review Comment:
`--maxSize` is enforced using the cumulative *input* file sizes (`LocatedFileStatus#getLen()`), not the actual size of the generated output ORC part files. The option description currently reads as if it caps the output file size; consider rewording to avoid implying a strict output-size limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4453487908
For the unaddressed comments, I believe they are minors which we can address later independently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "QianyongY (via GitHub)" <gi...@apache.org>.
QianyongY commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4456520928
Thanks @dongjoon-hyun and @cxzl25. The review items above were already fixed in the code—I had forgotten to mark some conversations as Resolved earlier, but I’ve done that now. Thanks again for your time and feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3119719128
##########
java/tools/src/test/org/apache/orc/tools/TestMergeFiles.java:
##########
@@ -107,4 +108,85 @@ public void testMerge() throws Exception {
assertEquals(10000 + 20000, reader.getNumberOfRows());
}
}
+
+ /**
+ * Verifies that --maxSize splits input files into multiple part files under the output
+ * directory. Three source files are created; a tight size threshold forces them to be
+ * written into at least two part files.
+ */
+ @Test
+ public void testMergeWithMaxSize() throws Exception {
Review Comment:
Could you add a test coverage for a single giant file which is bigger than `maxSize`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files [orc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2601:
URL: https://github.com/apache/orc/pull/2601#discussion_r3123026909
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +155,333 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes,
+ boolean overwrite) throws Exception {
+ DirMergeResult r = mergeBatchedIntoDir(conf, writerOptions, inputStatuses,
+ outputDir, maxSizeBytes, overwrite);
+
+ if (!r.unmergedFiles.isEmpty()) {
+ System.err.println("List of files that could not be merged:");
+ r.unmergedFiles.forEach(path -> System.err.println(path.toString()));
+ }
+
+ System.out.printf(
+ "Output path: %s, Input files size: %d, Merge files size: %d, Output files: %d%n",
+ outputDir, inputFiles.size(), r.mergedFileCount, r.partFileCount);
+ if (!r.unmergedFiles.isEmpty()) {
+ System.exit(1);
+ }
+ }
+
+ /**
+ * Preserve the relative directory structure of {@code inputRoot} under
+ * {@code outputRoot}: every directory containing ORC files (a "leaf") is merged
+ * independently, and its relative path is mirrored beneath {@code outputRoot}.
+ *
+ * <p>Leaves are located by a depth-first walk of {@code inputRoot}. A directory
+ * that contains both ORC files and subdirectories is considered ambiguous and
+ * raises an error. Hidden entries (names starting with {@code '_'} or
+ * {@code '.'}) are always skipped, matching Hive/Spark conventions for markers
+ * like {@code _SUCCESS}, {@code _committed_*} or {@code _temporary}.
+ *
+ * <p>If {@code --maxSize} was not supplied ({@code maxSizeBytes == 0}), each
+ * leaf is merged into a single {@code part-00000.orc} file; otherwise
+ * ({@code maxSizeBytes > 0}, already validated in {@link #main}) the files
+ * under a leaf are split into size-bounded part files exactly as in the flat
+ * multi-file mode.
+ */
+ private static void mergePreserveStructure(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ Path inputRoot,
+ Path outputRoot,
+ long maxSizeBytes,
+ boolean ignoreExtension,
+ boolean overwrite) throws Exception {
+ FileSystem inFs = inputRoot.getFileSystem(conf);
+ if (!inFs.exists(inputRoot) || !inFs.getFileStatus(inputRoot).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Input path must be an existing directory with --preserveStructure: " + inputRoot);
+ }
+
+ FileSystem outFs = outputRoot.getFileSystem(conf);
+ prepareOutputDir(outFs, outputRoot, overwrite, "--preserveStructure");
+
+ Path qualifiedInputRoot = inFs.makeQualified(inputRoot);
+ List<Path> leaves = new ArrayList<>();
+ collectLeafDirs(inFs, qualifiedInputRoot, ignoreExtension, leaves);
Review Comment:
In `--preserveStructure` mode, `prepareOutputDir` is invoked before validating/collecting leaf directories. Since `prepareOutputDir` may delete an existing non-empty output directory when `--overwrite` is set, an invalid input tree (eg a mixed directory) can still trigger destructive deletion before the tool fails. Consider collecting/validating leaves first (or at least delaying deletion until after validation succeeds), and only then preparing/overwriting the output directory.
##########
java/tools/src/java/org/apache/orc/tools/MergeFiles.java:
##########
@@ -100,11 +155,333 @@ public static void main(Configuration conf, String[] args) throws Exception {
}
}
+ /**
+ * Multi-output behavior when --maxSize is set.
+ * Input files are grouped by cumulative raw file size; each group is merged into
+ * a separate part file (part-00000.orc, part-00001.orc, ...) under outputDir.
+ * A single file whose size already exceeds maxSizeBytes is placed in its own part.
+ */
+ private static void mergeIntoMultipleFiles(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ List<Path> inputFiles,
+ Path outputDir,
+ long maxSizeBytes,
+ boolean overwrite) throws Exception {
+ DirMergeResult r = mergeBatchedIntoDir(conf, writerOptions, inputStatuses,
+ outputDir, maxSizeBytes, overwrite);
+
+ if (!r.unmergedFiles.isEmpty()) {
+ System.err.println("List of files that could not be merged:");
+ r.unmergedFiles.forEach(path -> System.err.println(path.toString()));
+ }
+
+ System.out.printf(
+ "Output path: %s, Input files size: %d, Merge files size: %d, Output files: %d%n",
+ outputDir, inputFiles.size(), r.mergedFileCount, r.partFileCount);
+ if (!r.unmergedFiles.isEmpty()) {
+ System.exit(1);
+ }
+ }
+
+ /**
+ * Preserve the relative directory structure of {@code inputRoot} under
+ * {@code outputRoot}: every directory containing ORC files (a "leaf") is merged
+ * independently, and its relative path is mirrored beneath {@code outputRoot}.
+ *
+ * <p>Leaves are located by a depth-first walk of {@code inputRoot}. A directory
+ * that contains both ORC files and subdirectories is considered ambiguous and
+ * raises an error. Hidden entries (names starting with {@code '_'} or
+ * {@code '.'}) are always skipped, matching Hive/Spark conventions for markers
+ * like {@code _SUCCESS}, {@code _committed_*} or {@code _temporary}.
+ *
+ * <p>If {@code --maxSize} was not supplied ({@code maxSizeBytes == 0}), each
+ * leaf is merged into a single {@code part-00000.orc} file; otherwise
+ * ({@code maxSizeBytes > 0}, already validated in {@link #main}) the files
+ * under a leaf are split into size-bounded part files exactly as in the flat
+ * multi-file mode.
+ */
+ private static void mergePreserveStructure(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ Path inputRoot,
+ Path outputRoot,
+ long maxSizeBytes,
+ boolean ignoreExtension,
+ boolean overwrite) throws Exception {
+ FileSystem inFs = inputRoot.getFileSystem(conf);
+ if (!inFs.exists(inputRoot) || !inFs.getFileStatus(inputRoot).isDirectory()) {
+ throw new IllegalArgumentException(
+ "Input path must be an existing directory with --preserveStructure: " + inputRoot);
+ }
+
+ FileSystem outFs = outputRoot.getFileSystem(conf);
+ prepareOutputDir(outFs, outputRoot, overwrite, "--preserveStructure");
+
+ Path qualifiedInputRoot = inFs.makeQualified(inputRoot);
+ List<Path> leaves = new ArrayList<>();
+ collectLeafDirs(inFs, qualifiedInputRoot, ignoreExtension, leaves);
+
+ if (leaves.isEmpty()) {
+ System.err.println("No leaf directories containing ORC files found under: " + inputRoot);
+ System.exit(1);
+ }
+
+ long effectiveMax = maxSizeBytes > 0 ? maxSizeBytes : Long.MAX_VALUE;
+ int totalInputFiles = 0;
+ int totalMerged = 0;
+ int totalPartFiles = 0;
+ List<Path> allUnmerged = new ArrayList<>();
+
+ for (Path leaf : leaves) {
+ String relative = relativize(qualifiedInputRoot, inFs.makeQualified(leaf));
+ Path outputLeaf = relative.isEmpty() ? outputRoot : new Path(outputRoot, relative);
+
+ List<LocatedFileStatus> leafStatuses =
+ listLeafFiles(inFs, leaf, ignoreExtension);
+ if (leafStatuses.isEmpty()) {
+ continue;
+ }
+
+ // Each leaf's output directory lives under a freshly-prepared outputRoot,
+ // so we don't need (and shouldn't force) overwrite semantics here.
+ DirMergeResult r = mergeBatchedIntoDir(
+ conf, writerOptions, leafStatuses, outputLeaf, effectiveMax, false);
+
+ totalInputFiles += leafStatuses.size();
+ totalMerged += r.mergedFileCount;
+ totalPartFiles += r.partFileCount;
+ allUnmerged.addAll(r.unmergedFiles);
+
+ System.out.printf(
+ "Leaf: %s -> %s, Input files: %d, Merge files: %d, Output files: %d%n",
+ leaf, outputLeaf, leafStatuses.size(), r.mergedFileCount, r.partFileCount);
+ }
+
+ if (!allUnmerged.isEmpty()) {
+ System.err.println("List of files that could not be merged:");
+ allUnmerged.forEach(path -> System.err.println(path.toString()));
+ }
+
+ System.out.printf(
+ "Output root: %s, Leaves: %d, Total input files: %d, "
+ + "Total merge files: %d, Total output files: %d%n",
+ outputRoot, leaves.size(), totalInputFiles, totalMerged, totalPartFiles);
+ if (!allUnmerged.isEmpty()) {
+ System.exit(1);
+ }
+ }
+
+ /**
+ * Core per-directory merge: groups {@code inputStatuses} into size-bounded batches
+ * and writes each batch as {@code part-NNNNN.orc} under {@code outputDir}. This
+ * helper contains no I/O side effects on stdout/stderr or process exit; callers
+ * aggregate/report results themselves.
+ */
+ private static DirMergeResult mergeBatchedIntoDir(Configuration conf,
+ OrcFile.WriterOptions writerOptions,
+ List<LocatedFileStatus> inputStatuses,
+ Path outputDir,
+ long maxSizeBytes,
+ boolean overwrite) throws Exception {
+ FileSystem outFs = outputDir.getFileSystem(conf);
+ prepareOutputDir(outFs, outputDir, overwrite, "multi-file merge");
+
+ List<List<Path>> batches = new ArrayList<>();
+ List<Path> currentBatch = new ArrayList<>();
+ long currentBatchSize = 0;
+
+ for (LocatedFileStatus status : inputStatuses) {
+ long fileSize = status.getLen();
+ if (!currentBatch.isEmpty() && currentBatchSize > maxSizeBytes - fileSize) {
+ batches.add(currentBatch);
+ currentBatch = new ArrayList<>();
+ currentBatchSize = 0;
+ }
+ currentBatch.add(status.getPath());
+ currentBatchSize += fileSize;
+ }
+ if (!currentBatch.isEmpty()) {
+ batches.add(currentBatch);
+ }
+
+ int totalMerged = 0;
+ List<Path> allUnmerged = new ArrayList<>();
+
+ for (int i = 0; i < batches.size(); i++) {
+ List<Path> batch = batches.get(i);
+ Path partOutput = new Path(outputDir, String.format(PART_FILE_FORMAT, i));
+ List<Path> merged = OrcFile.mergeFiles(partOutput, writerOptions.clone(), batch);
+ totalMerged += merged.size();
+
+ if (merged.size() != batch.size()) {
+ Set<Path> mergedSet = new HashSet<>(merged);
+ for (Path p : batch) {
+ if (!mergedSet.contains(p)) {
+ allUnmerged.add(p);
+ }
+ }
+ }
+ }
+
+ return new DirMergeResult(batches.size(), totalMerged, allUnmerged);
+ }
Review Comment:
`mergeBatchedIntoDir` reports `partFileCount` as `batches.size()` and uses the batch index for the part filename. If `OrcFile.mergeFiles(...)` returns an empty list for a batch (eg all inputs are skipped as non-ORC/corrupt when `--ignoreExtension` is used), no part file is created, causing gaps like missing `part-00000.orc` and an incorrect output-file count in the summary. Track the number of actually written part files (and use a separate part index that only increments when at least one file is merged) so the part numbering/count matches the filesystem state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ORC-2149: Supports merging multiple ORC files with the same schema into multiple ORC files. [orc]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #2601:
URL: https://github.com/apache/orc/pull/2601#issuecomment-4254794017
Thank you for making a PR, @QianyongY .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@orc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org