This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new cc5c6a1223 Issue #2251 : Fix integer overflow in ParquetInput for
files larger than 2GB (#7694)
cc5c6a1223 is described below
commit cc5c6a1223f55a13eaad85400037232e673f6671
Author: hma (Atol CD) <[email protected]>
AuthorDate: Thu Jul 30 16:54:47 2026 +0200
Issue #2251 : Fix integer overflow in ParquetInput for files larger than
2GB (#7694)
---
.../hop/parquet/transforms/input/ParquetInput.java | 15 +--
.../parquet/transforms/input/ParquetInputData.java | 3 +-
.../parquet/transforms/input/ParquetInputMeta.java | 11 +-
.../parquet/transforms/input/ParquetStream.java | 131 +++++++++++++++++----
.../parquet/transforms/output/ParquetTestUtil.java | 11 +-
5 files changed, 122 insertions(+), 49 deletions(-)
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInput.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInput.java
index 2352ad73fd..d51145e113 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInput.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInput.java
@@ -17,11 +17,9 @@
package org.apache.hop.parquet.transforms.input;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import org.apache.commons.io.IOUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.exception.HopException;
@@ -118,13 +116,8 @@ public class ParquetInput extends
BaseTransform<ParquetInputMeta, ParquetInputDa
// optional lineage
}
}
- data.inputStream = HopVfs.getInputStream(fileObject);
- // Reads the whole file into memory...
- //
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream((int)
size);
- IOUtils.copy(data.inputStream, outputStream);
- ParquetStream inputFile = new ParquetStream(outputStream.toByteArray(),
filename);
+ ParquetStream inputFile = new ParquetStream(fileObject, filename);
ParquetReadSupport readSupport = new ParquetReadSupport(fields);
data.reader = new ParquetReaderBuilder<>(readSupport, inputFile).build();
@@ -146,10 +139,12 @@ public class ParquetInput extends
BaseTransform<ParquetInputMeta, ParquetInputDa
}
public void closeFile() {
- if (!data.readerClosed && data.reader != null && data.inputStream != null)
{
+ if (!data.readerClosed && data.reader != null) {
try {
data.reader.close();
- data.inputStream.close();
+ if (data.parquetStream != null) {
+ data.parquetStream.close();
+ }
} catch (IOException e) {
logError("Unable to properly close parquet reader!");
}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputData.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputData.java
index 2ebf3aae4f..23736e32ee 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputData.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputData.java
@@ -17,7 +17,6 @@
package org.apache.hop.parquet.transforms.input;
-import java.io.InputStream;
import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.pipeline.transform.BaseTransformData;
@@ -30,7 +29,7 @@ public class ParquetInputData extends BaseTransformData
implements ITransformDat
public IRowMeta outputRowMeta;
public int filenameFieldIndex;
public ParquetReader<RowMetaAndData> reader;
- public InputStream inputStream;
+ public ParquetStream parquetStream;
public boolean readerClosed = false;
public ParquetInputData() {
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputMeta.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputMeta.java
index 1e6c439a40..5d922a0d63 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputMeta.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetInputMeta.java
@@ -17,13 +17,10 @@
package org.apache.hop.parquet.transforms.input;
-import java.io.ByteArrayOutputStream;
-import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
-import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.RowMetaAndData;
@@ -120,14 +117,8 @@ public class ParquetInputMeta extends
BaseTransformMeta<ParquetInput, ParquetInp
try {
FileObject fileObject =
HopVfs.getFileObject(variables.resolve(filename), variables);
- long size = fileObject.getContent().getSize();
- InputStream inputStream = HopVfs.getInputStream(fileObject);
+ ParquetStream inputFile = new ParquetStream(fileObject, filename);
- // Reads the whole file into memory...
- //
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream((int)
size);
- IOUtils.copy(inputStream, outputStream);
- ParquetStream inputFile = new ParquetStream(outputStream.toByteArray(),
filename);
// Empty list of fields to retrieve: we still grab the schema
//
ParquetReadSupport readSupport = new ParquetReadSupport(new
ArrayList<>());
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetStream.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetStream.java
index 1719bd7676..d5de7cbb12 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetStream.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetStream.java
@@ -17,39 +17,68 @@
package org.apache.hop.parquet.transforms.input;
-import java.io.ByteArrayInputStream;
+import java.io.Closeable;
import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Paths;
+import org.apache.commons.vfs2.FileObject;
+import org.apache.hop.core.vfs.HopVfs;
import org.apache.parquet.io.DelegatingSeekableInputStream;
import org.apache.parquet.io.InputFile;
+import org.apache.parquet.io.LocalInputFile;
import org.apache.parquet.io.SeekableInputStream;
-public class ParquetStream implements InputFile {
- private final byte[] data;
+public class ParquetStream implements InputFile, Closeable {
+
+ private final FileObject fileObject;
private final String filename;
+ private final boolean isLocal;
+ private final LocalInputFile localInputFile;
- public ParquetStream(byte[] contents, String filename) {
+ public ParquetStream(FileObject fileObject, String filename) throws
IOException {
+ this.fileObject = fileObject;
this.filename = filename;
- this.data = contents;
+ // Detect if the file is local by checking the VFS scheme
+ // For remote files, localInputFile is null. VfsSeekableInputStream will
be used instead
+ this.isLocal = "file".equals(fileObject.getName().getScheme());
+ this.localInputFile =
+ isLocal ? new
LocalInputFile(Paths.get(fileObject.getName().getPath())) : null;
}
@Override
public long getLength() throws IOException {
- return this.data.length;
+ if (isLocal) {
+ return localInputFile.getLength();
+ }
+ return fileObject.getContent().getSize();
}
@Override
public SeekableInputStream newStream() throws IOException {
- return new DelegatingSeekableInputStream(new
SeekableByteArrayInputStream(this.data)) {
- @Override
- public void seek(long newPos) throws IOException {
- ((SeekableByteArrayInputStream) this.getStream()).setPos((int) newPos);
- }
+ if (isLocal) {
+ // Native Parquet implementation
+ return localInputFile.newStream();
+ } else {
+ // For remote file, DelegatingSeekableInputStream handles basic
read/skip operations
+ return new DelegatingSeekableInputStream(new
VfsSeekableInputStream(fileObject)) {
+ @Override
+ public void seek(long newPos) throws IOException {
+ ((VfsSeekableInputStream) getStream()).seek(newPos);
+ }
- @Override
- public long getPos() throws IOException {
- return ((SeekableByteArrayInputStream) this.getStream()).getPos();
- }
- };
+ @Override
+ public long getPos() throws IOException {
+ return ((VfsSeekableInputStream) getStream()).getPos();
+ }
+ };
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!isLocal && fileObject != null) {
+ fileObject.getContent().close();
+ }
}
@Override
@@ -57,17 +86,73 @@ public class ParquetStream implements InputFile {
return "ParquetStream of file '" + filename + "'";
}
- private static class SeekableByteArrayInputStream extends
ByteArrayInputStream {
- public SeekableByteArrayInputStream(byte[] buf) {
- super(buf);
+ // SeekableInputStream implementation for remote files
+ private static class VfsSeekableInputStream extends InputStream {
+
+ private final FileObject fileObject;
+ private InputStream currentStream;
+ private long pos = 0;
+
+ public VfsSeekableInputStream(FileObject fileObject) throws IOException {
+ this.fileObject = fileObject;
+ this.currentStream = HopVfs.getInputStream(fileObject);
}
- public void setPos(int pos) {
- this.pos = pos;
+ // Reads a single byte and advances the position by 1
+ @Override
+ public int read() throws IOException {
+ int b = currentStream.read();
+ if (b != -1) {
+ pos++;
+ }
+
+ return b;
+ }
+
+ // Reads up to len bytes into buffer b starting at offset off
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ int n = currentStream.read(b, off, len);
+ if (n > 0) {
+ pos += n;
+ }
+
+ return n;
+ }
+
+ public long getPos() {
+ return pos;
+ }
+
+ public void seek(long newPos) throws IOException {
+ currentStream.close();
+ currentStream = HopVfs.getInputStream(fileObject);
+ pos = 0;
+ long remaining = newPos;
+ while (remaining > 0) {
+ long skipped = currentStream.skip(remaining);
+ if (skipped > 0) {
+ // Advance by the number of bytes skipped
+ remaining -= skipped;
+ // If skip() returns 0, stream may be blocked
+ } else if (skipped == 0) {
+ // Try reading one byte manually to advance
+ if (currentStream.read() == -1) {
+ break;
+ }
+ // One byte consumed via read(), decrement remaining accordingly
+ remaining--;
+ } else {
+ break;
+ }
+ }
+ // Update position to reflect where we actually ended up
+ pos = newPos - remaining;
}
- public int getPos() {
- return this.pos;
+ @Override
+ public void close() throws IOException {
+ currentStream.close();
}
}
}
diff --git
a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetTestUtil.java
b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetTestUtil.java
index 5cc01b5642..45fd8c19c6 100644
---
a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetTestUtil.java
+++
b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetTestUtil.java
@@ -19,17 +19,19 @@ package org.apache.hop.parquet.transforms.output;
import java.io.IOException;
import java.net.URISyntaxException;
-import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
+import org.apache.commons.vfs2.FileObject;
import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopFileException;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.core.variables.Variables;
+import org.apache.hop.core.vfs.HopVfs;
import org.apache.hop.parquet.transforms.input.ParquetField;
import org.apache.hop.parquet.transforms.input.ParquetInputMeta;
import org.apache.hop.parquet.transforms.input.ParquetReadSupport;
@@ -76,9 +78,10 @@ final class ParquetTestUtil {
}
static List<RowMetaAndData> readAllRows(String filename, List<ParquetField>
fields)
- throws IOException {
- byte[] data = Files.readAllBytes(Path.of(filename));
- ParquetStream inputFile = new ParquetStream(data, filename);
+ throws IOException, HopFileException {
+ FileObject fileObject = HopVfs.getFileObject(filename);
+ ParquetStream inputFile = new ParquetStream(fileObject, filename);
+
ParquetReadSupport readSupport = new ParquetReadSupport(fields);
try (ParquetReader<RowMetaAndData> reader =
new ParquetReaderBuilder<>(readSupport, inputFile).build()) {
