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 0a6e08ad24 Issue #8080 : Support Hive-style partitioning in the
Parquet File Output transform (#8254)
0a6e08ad24 is described below
commit 0a6e08ad243f6518652e92898a3eaed9e4dd00c4
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Fri Sep 4 06:26:51 2026 -0400
Issue #8080 : Support Hive-style partitioning in the Parquet File Output
transform (#8254)
* Issue #8080 : Support Hive-style partitioning in the Parquet File Output
transform
Data partitioning was only available in Native Spark mode. The standard
Parquet File Output transform can now partition by one or more incoming
fields, writing the Hive-style layout that Spark's partitionBy produces:
/datalake/sales/year=2026/month=08/part-00-0000-3f2a1b9c.parquet
The partition fields are not written into the files, matching partitionBy
and what a reader expects when recovering the column from the path. Nulls
become __HIVE_DEFAULT_PARTITION__, and values are percent-escaped so a
value containing a separator stays one folder level.
Four write modes decide what happens to data already in a partition
folder: append (unchanged behaviour), overwrite partitions, fail if
exists, and overwrite all. Overwrite-partitions clears a folder once per
run rather than once per file, so a reopened partition cannot delete rows
the same run just wrote.
Every open Parquet writer buffers up to a full row group, so a wide
partition key would otherwise exhaust memory. At most "maximum open
partitions" writers are held open and the least recently written one is
closed when another is needed.
With no partition fields configured the transform behaves exactly as
before; the partitioned path is a separate branch in processRow.
Also fixes a missing '=' in messages_en_US.properties that left the
row group size dialog label unresolved.
* Issue #8080 : Use Lombok on ParquetPartitionField and strip trailing
backslash in partitionFolder
---------
Co-authored-by: mattcasters <[email protected]>
---
.../parquet/transforms/output/ParquetOutput.java | 494 ++++++++++++++++++---
.../transforms/output/ParquetOutputData.java | 62 +++
.../transforms/output/ParquetOutputDialog.java | 113 +++++
.../transforms/output/ParquetOutputMeta.java | 86 ++++
.../transforms/output/ParquetPartitionField.java | 43 ++
.../transforms/output/ParquetWriteMode.java | 75 ++++
.../output/messages/messages_en_US.properties | 17 +-
.../output/ParquetOutputPartitionTest.java | 419 +++++++++++++++++
8 files changed, 1252 insertions(+), 57 deletions(-)
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutput.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutput.java
index 9c7fef4f21..be3e43be82 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutput.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutput.java
@@ -17,15 +17,22 @@
package org.apache.hop.parquet.transforms.output;
+import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.commons.vfs2.FileObject;
+import org.apache.commons.vfs2.Selectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hop.core.Const;
import org.apache.hop.core.RowMetaAndData;
@@ -49,6 +56,15 @@ import org.apache.parquet.schema.MessageType;
public class ParquetOutput extends BaseTransform<ParquetOutputMeta,
ParquetOutputData> {
+ /** How many partitions are written to at the same time when nothing else is
configured. */
+ static final int DEFAULT_MAX_OPEN_PARTITIONS = 10;
+
+ /**
+ * The partition value Hive and Spark both use for a null, so that
partitioned data written here
+ * can be read back by them.
+ */
+ static final String DEFAULT_PARTITION_NAME = "__HIVE_DEFAULT_PARTITION__";
+
public ParquetOutput(
TransformMeta transformMeta,
ParquetOutputMeta meta,
@@ -73,6 +89,11 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
Const.toIntExpanded(
resolve(meta.getRowGroupSize()),
ParquetProperties.DEFAULT_PAGE_ROW_COUNT_LIMIT);
data.maxSplitSizeRows =
Const.toLongExpanded(resolve(meta.getFileSplitSize()), -1);
+ data.maxOpenPartitions =
+ Const.toIntExpanded(resolve(meta.getMaxOpenPartitions()),
DEFAULT_MAX_OPEN_PARTITIONS);
+ if (data.maxOpenPartitions < 1) {
+ data.maxOpenPartitions = 1;
+ }
return super.init();
}
@@ -87,7 +108,11 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
}
if (row == null) {
- closeFile();
+ if (meta.isPartitioning()) {
+ closeAllPartitionWriters();
+ } else {
+ closeFile();
+ }
setOutputDone();
return false;
}
@@ -95,12 +120,24 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
if (first) {
first = false;
resolveOutputFields();
- openNewFile();
+ if (meta.isPartitioning()) {
+ data.runToken = UUID.randomUUID().toString().substring(0, 8);
+ initWriterProperties();
+ data.messageType = buildSchema();
+ data.partitionWriters = data.newPartitionWriterMap();
+ data.clearedPartitions = new HashSet<>();
+ if (meta.getWriteMode() == ParquetWriteMode.OverwriteAll) {
+ clearBaseFolder();
+ }
+ } else {
+ openNewFile();
+ }
}
// See if we don't need to create a new file split into parts...
//
- if (meta.isFilenameIncludingSplitNr()
+ if (!meta.isPartitioning()
+ && meta.isFilenameIncludingSplitNr()
&& data.maxSplitSizeRows > 0
&& data.splitRowCount >= data.maxSplitSizeRows) {
// Close file and start a new one...
@@ -145,9 +182,16 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
}
}
- data.writer.write(new RowMetaAndData(parquetRowMeta, parquetRow));
+ if (meta.isPartitioning()) {
+ ParquetOutputData.PartitionWriter partitionWriter =
+ getPartitionWriter(partitionPath(getInputRowMeta(), row));
+ partitionWriter.writer.write(new RowMetaAndData(parquetRowMeta,
parquetRow));
+ partitionWriter.rowCount++;
+ } else {
+ data.writer.write(new RowMetaAndData(parquetRowMeta, parquetRow));
+ data.splitRowCount++;
+ }
incrementLinesOutput();
- data.splitRowCount++;
} catch (Exception e) {
throw new HopException("Error writing row to parquet file", e);
}
@@ -160,12 +204,60 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
data.splitRowCount = 0;
data.split++;
- // Hadoop configuration
+ initWriterProperties();
+
+ MessageType messageType = buildSchema();
+
+ // Calculate the filename...
//
+ data.filename = buildFilename(getPipeline().getExecutionStartDate());
+
+ try {
+ FileObject fileObject = HopVfs.getFileObject(data.filename, variables);
+
+ // See if we need to create the parent folder(s)...
+ //
+ if (meta.isFilenameCreatingParentFolders()) {
+ FileObject parentFolder = fileObject.getParent();
+ if (parentFolder != null && !parentFolder.exists()) {
+ // Try to create the parent folder...
+ //
+ parentFolder.createFolder();
+ }
+ }
+
+ data.outputStream = HopVfs.getOutputStream(data.filename, false,
variables);
+ data.countingStream = new CountingOutputStream(data.outputStream);
+ data.outputFile = new ParquetOutputFile(data.countingStream);
+
+ data.writer =
+ new ParquetWriterBuilder(
+ messageType,
+ data.avroSchema,
+ data.outputFile,
+ data.sourceFieldIndexes,
+ data.outputFields)
+ .withPageSize(data.pageSize)
+ .withDictionaryPageSize(data.dictionaryPageSize)
+ .withValidation(ParquetWriter.DEFAULT_IS_VALIDATING_ENABLED)
+ .withCompressionCodec(meta.getCompressionCodec())
+ .withRowGroupSize(data.rowGroupSize)
+ .withWriterVersion(data.props.getWriterVersion())
+ .withWriteMode(ParquetFileWriter.Mode.CREATE)
+ .build();
+
+ } catch (Exception e) {
+ throw new HopException("Unable to create output file '" + data.filename
+ "'", e);
+ }
+ }
+
+ /**
+ * Sets up the Hadoop configuration and Parquet properties. Both the
single-file and the
+ * partitioned paths need these before a writer can be built.
+ */
+ private void initWriterProperties() {
data.conf = new Configuration();
- // Parquet Properties
- //
ParquetProperties.Builder builder = ParquetProperties.builder();
builder =
switch (meta.getVersion()) {
@@ -173,7 +265,14 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
case Version2 ->
builder.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_2_0);
};
data.props = builder.build();
+ }
+ /**
+ * Builds the Avro schema for the resolved output fields and converts it to
a Parquet schema. Kept
+ * separate from opening a file so the partitioned path can build it once
and reuse it for every
+ * partition.
+ */
+ private MessageType buildSchema() throws HopException {
SchemaBuilder.FieldAssembler<Schema> fieldAssembler =
SchemaBuilder.record("ApacheHopParquetSchema").fields();
@@ -225,62 +324,28 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
// Convert from Avro to Parquet schema
//
- MessageType messageType = new
AvroSchemaConverter().convert(data.avroSchema);
-
- // Calculate the filename...
- //
- data.filename = buildFilename(getPipeline().getExecutionStartDate());
-
- try {
- FileObject fileObject = HopVfs.getFileObject(data.filename, variables);
-
- // See if we need to create the parent folder(s)...
- //
- if (meta.isFilenameCreatingParentFolders()) {
- FileObject parentFolder = fileObject.getParent();
- if (parentFolder != null && !parentFolder.exists()) {
- // Try to create the parent folder...
- //
- parentFolder.createFolder();
- }
- }
-
- data.outputStream = HopVfs.getOutputStream(data.filename, false,
variables);
- data.countingStream = new CountingOutputStream(data.outputStream);
- data.outputFile = new ParquetOutputFile(data.countingStream);
-
- data.writer =
- new ParquetWriterBuilder(
- messageType,
- data.avroSchema,
- data.outputFile,
- data.sourceFieldIndexes,
- data.outputFields)
- .withPageSize(data.pageSize)
- .withDictionaryPageSize(data.dictionaryPageSize)
- .withValidation(ParquetWriter.DEFAULT_IS_VALIDATING_ENABLED)
- .withCompressionCodec(meta.getCompressionCodec())
- .withRowGroupSize(data.rowGroupSize)
- .withWriterVersion(data.props.getWriterVersion())
- .withWriteMode(ParquetFileWriter.Mode.CREATE)
- .build();
-
- } catch (Exception e) {
- throw new HopException("Unable to create output file '" + data.filename
+ "'", e);
- }
+ return new AvroSchemaConverter().convert(data.avroSchema);
}
void resolveOutputFields() throws HopException {
data.outputFields = new ArrayList<>();
data.sourceFieldIndexes = new ArrayList<>();
+ resolvePartitionFields();
if (meta.getFields() == null || meta.getFields().isEmpty()) {
IRowMeta inputRowMeta = getInputRowMeta();
for (int i = 0; i < inputRowMeta.size(); i++) {
+ if (data.partitionFieldIndexes.contains(i)) {
+ // The value lives in the folder name, so it is not written into the
file. This is what
+ // Spark's partitionBy() does, and what a reader expects when it
recovers the column from
+ // the path.
+ continue;
+ }
String fieldName = inputRowMeta.getValueMeta(i).getName();
data.outputFields.add(new ParquetField(fieldName, fieldName));
data.sourceFieldIndexes.add(i);
}
+ verifyFieldsRemain();
return;
}
@@ -289,10 +354,44 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
if (index < 0) {
throw new HopException("Unable to find source field '" +
field.getSourceFieldName() + "'");
}
+ if (data.partitionFieldIndexes.contains(index)) {
+ continue;
+ }
String targetFieldName = Const.NVL(field.getTargetFieldName(),
field.getSourceFieldName());
data.outputFields.add(new ParquetField(field.getSourceFieldName(),
targetFieldName));
data.sourceFieldIndexes.add(index);
}
+ verifyFieldsRemain();
+ }
+
+ /** Resolves the configured partition field names to indexes in the incoming
row. */
+ private void resolvePartitionFields() throws HopException {
+ data.partitionFieldIndexes = new ArrayList<>();
+ if (!meta.isPartitioning()) {
+ return;
+ }
+ for (ParquetPartitionField field : meta.getPartitionFields()) {
+ String name = field.getName();
+ if (name == null || name.trim().isEmpty()) {
+ continue;
+ }
+ int index = getInputRowMeta().indexOfValue(name);
+ if (index < 0) {
+ throw new HopException("Unable to find partition field '" + name + "'
in the input");
+ }
+ if (data.partitionFieldIndexes.contains(index)) {
+ throw new HopException("Partition field '" + name + "' is listed more
than once");
+ }
+ data.partitionFieldIndexes.add(index);
+ }
+ }
+
+ private void verifyFieldsRemain() throws HopException {
+ if (data.outputFields.isEmpty()) {
+ throw new HopException(
+ "Every output field is a partition field, which would leave nothing
to write into the "
+ + "Parquet files. Leave at least one non-partition field.");
+ }
}
String buildFilename(Date date) {
@@ -330,6 +429,285 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
return filename;
}
+ /**
+ * Builds the {@code name=value/...} folder path for a row, in the
configured partition field
+ * order. Nulls become {@link #DEFAULT_PARTITION_NAME} and characters that
would break the path
+ * are percent-escaped, both so the result can be read back by Hive and
Spark.
+ */
+ String partitionPath(IRowMeta rowMeta, Object[] row) throws HopException {
+ StringBuilder path = new StringBuilder();
+ for (int i = 0; i < data.partitionFieldIndexes.size(); i++) {
+ int index = data.partitionFieldIndexes.get(i);
+ String value = rowMeta.getString(row, index);
+ if (i > 0) {
+ path.append('/');
+ }
+ path.append(escapePathValue(rowMeta.getValueMeta(index).getName()))
+ .append('=')
+ .append(
+ value == null || value.isEmpty() ? DEFAULT_PARTITION_NAME :
escapePathValue(value));
+ }
+ return path.toString();
+ }
+
+ /** Percent-escapes the characters that cannot appear in a partition folder
name. */
+ static String escapePathValue(String value) {
+ StringBuilder escaped = new StringBuilder(value.length());
+ for (int i = 0; i < value.length(); i++) {
+ char c = value.charAt(i);
+ boolean needsEscape =
+ c == '%' || c == '=' || c == '/' || c == '\\' || c == ':' || c ==
'"' || c == '\''
+ || c < 0x20 || c == 0x7f;
+ if (needsEscape) {
+ escaped.append('%').append(String.format("%02X", (int) c));
+ } else {
+ escaped.append(c);
+ }
+ }
+ return escaped.toString();
+ }
+
+ /**
+ * Returns the open writer for a partition, opening one if needed. When too
many partitions are
+ * open at once the least recently written one is closed first, so a wide
partition key cannot run
+ * the transform out of memory: every open Parquet writer buffers up to a
full row group.
+ */
+ private ParquetOutputData.PartitionWriter getPartitionWriter(String
partitionPath)
+ throws HopException {
+ ParquetOutputData.PartitionWriter partitionWriter =
data.partitionWriters.get(partitionPath);
+
+ if (partitionWriter != null
+ && meta.isFilenameIncludingSplitNr()
+ && data.maxSplitSizeRows > 0
+ && partitionWriter.rowCount >= data.maxSplitSizeRows) {
+ // This partition's current file is full, roll over to the next part.
+ closePartitionWriter(partitionPath, partitionWriter);
+ data.partitionWriters.remove(partitionPath);
+ partitionWriter = null;
+ }
+
+ if (partitionWriter != null) {
+ return partitionWriter;
+ }
+
+ while (data.partitionWriters.size() >= data.maxOpenPartitions) {
+ Iterator<Map.Entry<String, ParquetOutputData.PartitionWriter>> iterator =
+ data.partitionWriters.entrySet().iterator();
+ Map.Entry<String, ParquetOutputData.PartitionWriter> oldest =
iterator.next();
+ closePartitionWriter(oldest.getKey(), oldest.getValue());
+ iterator.remove();
+ }
+
+ partitionWriter = openPartitionWriter(partitionPath);
+ data.partitionWriters.put(partitionPath, partitionWriter);
+ return partitionWriter;
+ }
+
+ private ParquetOutputData.PartitionWriter openPartitionWriter(String
partitionPath)
+ throws HopException {
+ String folder = partitionFolder(partitionPath);
+ applyWriteMode(partitionPath, folder);
+
+ String filename = buildPartitionFilename(folder,
getPipeline().getExecutionStartDate());
+ try {
+ FileObject fileObject = HopVfs.getFileObject(filename, variables);
+ FileObject parentFolder = fileObject.getParent();
+ if (parentFolder != null && !parentFolder.exists()) {
+ // Partition folders are created regardless of the "create parent
folders" option: the
+ // layout is what the transform was asked to produce, not something
the user typed.
+ parentFolder.createFolder();
+ }
+
+ OutputStream outputStream = HopVfs.getOutputStream(filename, false,
variables);
+ CountingOutputStream countingStream = new
CountingOutputStream(outputStream);
+ ParquetOutputFile outputFile = new ParquetOutputFile(countingStream);
+
+ ParquetWriter<RowMetaAndData> writer =
+ new ParquetWriterBuilder(
+ data.messageType,
+ data.avroSchema,
+ outputFile,
+ data.sourceFieldIndexes,
+ data.outputFields)
+ .withPageSize(data.pageSize)
+ .withDictionaryPageSize(data.dictionaryPageSize)
+ .withValidation(ParquetWriter.DEFAULT_IS_VALIDATING_ENABLED)
+ .withCompressionCodec(meta.getCompressionCodec())
+ .withRowGroupSize(data.rowGroupSize)
+ .withWriterVersion(data.props.getWriterVersion())
+ .withWriteMode(ParquetFileWriter.Mode.CREATE)
+ .build();
+
+ if (isDetailed()) {
+ logDetailed("Opened partition file '" + filename + "'");
+ }
+ return new ParquetOutputData.PartitionWriter(
+ filename, outputStream, countingStream, outputFile, writer);
+ } catch (Exception e) {
+ throw new HopException("Unable to create partition output file '" +
filename + "'", e);
+ }
+ }
+
+ /** The folder a partition's files go into: the base folder plus the
partition path. */
+ String partitionFolder(String partitionPath) {
+ String base = resolve(meta.getFilenameBase());
+ if (base.endsWith("/") || base.endsWith("\\")) {
+ base = base.substring(0, base.length() - 1);
+ }
+ return base + "/" + toVfsPath(partitionPath);
+ }
+
+ /**
+ * HopVfs resolves a path as a URI and percent-decodes it, so a {@code %2F}
we wrote for a value
+ * containing a separator would come back as a real folder level. Encoding
our own {@code %} as
+ * {@code %25} means VFS decodes it back to a literal {@code %}, leaving the
Hive-style {@code
+ * name=EU%2FWest} on disk for a reader to decode.
+ */
+ static String toVfsPath(String partitionPath) {
+ return partitionPath.replace("%", "%25");
+ }
+
+ /**
+ * Names a file inside a partition folder. The base name is not repeated,
since the folder already
+ * identifies the data; the copy number and a per-run sequence keep parallel
copies and re-opened
+ * partitions from colliding.
+ */
+ private String buildPartitionFilename(String folder, Date date) {
+ StringBuilder filename = new StringBuilder(folder).append("/part");
+ if (meta.isFilenameIncludingDate()) {
+ filename.append('-').append(new
SimpleDateFormat("yyyyMMdd").format(date));
+ }
+ if (meta.isFilenameIncludingTime()) {
+ filename.append('-').append(new SimpleDateFormat("HHmmss").format(date));
+ }
+ if (meta.isFilenameIncludingDateTime()) {
+ filename
+ .append('-')
+ .append(new
SimpleDateFormat(resolve(meta.getFilenameDateTimeFormat())).format(date));
+ }
+ filename.append('-').append(new DecimalFormat("00").format(getCopyNr()));
+ filename.append('-').append(new
DecimalFormat("0000").format(data.partitionFileNr++));
+ // Without something unique per run, a second run in append mode would
write the same
+ // part name and silently replace the first run's file.
+ filename.append('-').append(data.runToken);
+ if (data.isBeamContext()) {
+
filename.append('_').append(getLogChannelId()).append('_').append(data.getBeamBundleNr());
+ }
+ String extension = Const.NVL(resolve(meta.getFilenameExtension()),
"parquet");
+ String compressionExtension = meta.getCompressionCodec().getExtension();
+ if (meta.isFilenameCompressionBeforeExtension()) {
+ filename.append(compressionExtension).append('.').append(extension);
+ } else {
+ filename.append('.').append(extension).append(compressionExtension);
+ }
+ return filename.toString();
+ }
+
+ /**
+ * Applies the configured write mode the first time this run touches a
partition folder. The
+ * overwrite-all mode is handled once up front, before any file is opened.
+ */
+ private void applyWriteMode(String partitionPath, String folder) throws
HopException {
+ ParquetWriteMode mode = meta.getWriteMode();
+ if (mode == null || mode == ParquetWriteMode.Append || mode ==
ParquetWriteMode.OverwriteAll) {
+ return;
+ }
+ if (data.clearedPartitions.contains(partitionPath)) {
+ // Already dealt with; a re-opened partition must not wipe what this run
just wrote.
+ return;
+ }
+ data.clearedPartitions.add(partitionPath);
+
+ try {
+ FileObject folderObject = HopVfs.getFileObject(folder, variables);
+ if (!folderObject.exists()) {
+ return;
+ }
+ if (mode == ParquetWriteMode.FailIfExists) {
+ throw new HopException("Partition folder '" + folder + "' already
exists");
+ }
+ // OverwritePartitions
+ int deleted = folderObject.delete(Selectors.EXCLUDE_SELF);
+ if (isDetailed()) {
+ logDetailed("Emptied partition folder '" + folder + "', removed " +
deleted + " file(s)");
+ }
+ } catch (HopException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new HopException("Unable to apply write mode to partition folder
'" + folder + "'", e);
+ }
+ }
+
+ /** Empties the base folder once, for the overwrite-all mode. */
+ private void clearBaseFolder() throws HopException {
+ if (data.baseFolderCleared) {
+ return;
+ }
+ data.baseFolderCleared = true;
+ String base = partitionFolder("");
+ try {
+ FileObject folderObject = HopVfs.getFileObject(base, variables);
+ if (folderObject.exists()) {
+ int deleted = folderObject.delete(Selectors.EXCLUDE_SELF);
+ logBasic("Emptied output folder '" + base + "', removed " + deleted +
" file(s)");
+ }
+ } catch (Exception e) {
+ throw new HopException("Unable to empty output folder '" + base + "'",
e);
+ }
+ }
+
+ private void closeAllPartitionWriters() throws HopException {
+ if (data.partitionWriters == null) {
+ return;
+ }
+ // Copy first: closing reports lineage, and the map must not be mutated
while iterating.
+ List<Map.Entry<String, ParquetOutputData.PartitionWriter>> open =
+ new ArrayList<>(data.partitionWriters.entrySet());
+ data.partitionWriters.clear();
+ HopException firstFailure = null;
+ for (Map.Entry<String, ParquetOutputData.PartitionWriter> entry : open) {
+ try {
+ closePartitionWriter(entry.getKey(), entry.getValue());
+ } catch (HopException e) {
+ // Keep closing the rest so no file is left half-written, then report
the first failure.
+ if (firstFailure == null) {
+ firstFailure = e;
+ }
+ }
+ }
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ }
+
+ private void closePartitionWriter(
+ String partitionPath, ParquetOutputData.PartitionWriter partitionWriter)
throws HopException {
+ try {
+ partitionWriter.writer.close();
+ if (partitionWriter.countingStream != null) {
+ long written = partitionWriter.countingStream.getCount();
+ dataVolumeOut = (dataVolumeOut != null ? dataVolumeOut : 0L) + written;
+ if (!data.isBeamContext() && written > 0) {
+ try {
+ FileObject outFile =
HopVfs.getFileObject(partitionWriter.filename, variables);
+ LineageFileIoEmitter.emitTransformFileIo(
+ this, FileIoOperation.WRITE, null, outFile, written, true,
null);
+ } catch (Exception ignored) {
+ // optional lineage
+ }
+ }
+ }
+ } catch (Exception e) {
+ throw new HopException(
+ "Error closing file "
+ + partitionWriter.filename
+ + " for partition '"
+ + partitionPath
+ + "'",
+ e);
+ }
+ }
+
private void closeFile() throws HopException {
try {
data.writer.close();
@@ -354,19 +732,27 @@ public class ParquetOutput extends
BaseTransform<ParquetOutputMeta, ParquetOutpu
@Override
public void batchComplete() throws HopException {
if (!data.isBeamContext()) {
- closeFile();
+ if (meta.isPartitioning()) {
+ closeAllPartitionWriters();
+ } else {
+ closeFile();
+ }
}
}
@Override
public void startBundle() throws HopException {
- if (!first) {
+ if (!first && !meta.isPartitioning()) {
openNewFile();
}
}
@Override
public void finishBundle() throws HopException {
- closeFile();
+ if (meta.isPartitioning()) {
+ closeAllPartitionWriters();
+ } else {
+ closeFile();
+ }
}
}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputData.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputData.java
index ce3573cc72..cb58751034 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputData.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputData.java
@@ -18,7 +18,10 @@
package org.apache.hop.parquet.transforms.output;
import java.io.OutputStream;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hop.core.RowMetaAndData;
@@ -27,6 +30,7 @@ import org.apache.hop.pipeline.transform.BaseTransformData;
import org.apache.hop.pipeline.transform.ITransformData;
import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.schema.MessageType;
@SuppressWarnings("java:S1104")
public class ParquetOutputData extends BaseTransformData implements
ITransformData {
@@ -47,7 +51,65 @@ public class ParquetOutputData extends BaseTransformData
implements ITransformDa
public int dictionaryPageSize;
public Schema avroSchema;
+ /** The Parquet schema, built once and shared by every partition file. */
+ public MessageType messageType;
+
+ /** Indexes of the input fields the output is partitioned by, in the
configured order. */
+ public List<Integer> partitionFieldIndexes;
+
+ /**
+ * The writer per partition folder, keyed by the relative {@code
name=value/...} path. Access
+ * order is maintained so the least recently written partition can be closed
first when {@link
+ * #maxOpenPartitions} is reached.
+ */
+ public Map<String, PartitionWriter> partitionWriters;
+
+ /** Partition folders already emptied during this run, for the
overwrite-partitions mode. */
+ public Set<String> clearedPartitions;
+
+ /** Whether the base folder has been emptied yet, for the overwrite-all
mode. */
+ public boolean baseFolderCleared;
+
+ /** How many partitions may be written to at the same time. */
+ public int maxOpenPartitions;
+
+ /** Number of partition files opened so far, used to keep their names
unique. */
+ public int partitionFileNr;
+
+ /** Short token unique to this run, so a later run does not overwrite this
one's files. */
+ public String runToken;
+
public ParquetOutputData() {
super();
}
+
+ /** Everything needed to write, measure and close one partition folder's
current file. */
+ public static class PartitionWriter {
+ public String filename;
+ public OutputStream outputStream;
+ public CountingOutputStream countingStream;
+ public ParquetOutputFile outputFile;
+ public ParquetWriter<RowMetaAndData> writer;
+ public long rowCount;
+
+ public PartitionWriter(
+ String filename,
+ OutputStream outputStream,
+ CountingOutputStream countingStream,
+ ParquetOutputFile outputFile,
+ ParquetWriter<RowMetaAndData> writer) {
+ this.filename = filename;
+ this.outputStream = outputStream;
+ this.countingStream = countingStream;
+ this.outputFile = outputFile;
+ this.writer = writer;
+ }
+ }
+
+ /**
+ * Creates the partition writer map with access-order iteration for
least-recently-used eviction.
+ */
+ public Map<String, PartitionWriter> newPartitionWriterMap() {
+ return new LinkedHashMap<>(16, 0.75f, true);
+ }
}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputDialog.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputDialog.java
index fd6d570618..91869d3b39 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputDialog.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputDialog.java
@@ -69,6 +69,11 @@ public class ParquetOutputDialog extends BaseTransformDialog
{
private TextVar wDataPageSize;
private TextVar wDictionaryPageSize;
private TableView wFields;
+ private TableView wPartitionFields;
+ private Combo wWriteMode;
+ private TextVar wMaxOpenPartitions;
+ private Label wlWriteMode;
+ private Label wlMaxOpenPartitions;
private String returnValue;
@@ -424,6 +429,87 @@ public class ParquetOutputDialog extends
BaseTransformDialog {
wDictionaryPageSize.setLayoutData(fdDictionaryPageSize);
lastControl = wDictionaryPageSize;
+ Label wlPartitionFields = new Label(shell, SWT.LEFT);
+ wlPartitionFields.setText(
+ BaseMessages.getString(PKG,
"ParquetOutputDialog.PartitionFields.Label"));
+ PropsUi.setLook(wlPartitionFields);
+ FormData fdlPartitionFields = new FormData();
+ fdlPartitionFields.left = new FormAttachment(0, 0);
+ fdlPartitionFields.right = new FormAttachment(100, 0);
+ fdlPartitionFields.top = new FormAttachment(lastControl, margin);
+ wlPartitionFields.setLayoutData(fdlPartitionFields);
+
+ ColumnInfo[] partitionColumns =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG,
"ParquetOutputDialog.PartitionFieldsColumn.Field.Label"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ new String[0]),
+ };
+ wPartitionFields =
+ new TableView(
+ variables,
+ shell,
+ SWT.BORDER,
+ partitionColumns,
+ input.getPartitionFields().size(),
+ false,
+ null,
+ props);
+ PropsUi.setLook(wPartitionFields);
+ FormData fdPartitionFields = new FormData();
+ fdPartitionFields.left = new FormAttachment(0, 0);
+ fdPartitionFields.top = new FormAttachment(wlPartitionFields, margin);
+ fdPartitionFields.right = new FormAttachment(100, 0);
+ fdPartitionFields.height = (int) (100 * props.getZoomFactor());
+ wPartitionFields.setLayoutData(fdPartitionFields);
+ wPartitionFields.addModifyListener(e -> enableFields());
+ lastControl = wPartitionFields;
+
+ wlWriteMode = new Label(shell, SWT.RIGHT);
+ wlWriteMode.setText(BaseMessages.getString(PKG,
"ParquetOutputDialog.WriteMode.Label"));
+ wlWriteMode.setToolTipText(
+ BaseMessages.getString(PKG, "ParquetOutputDialog.WriteMode.ToolTip"));
+ PropsUi.setLook(wlWriteMode);
+ FormData fdlWriteMode = new FormData();
+ fdlWriteMode.left = new FormAttachment(0, 0);
+ fdlWriteMode.right = new FormAttachment(middle, -margin);
+ fdlWriteMode.top = new FormAttachment(lastControl, margin);
+ wlWriteMode.setLayoutData(fdlWriteMode);
+ wWriteMode = new Combo(shell, SWT.SINGLE | SWT.LEFT | SWT.BORDER |
SWT.READ_ONLY);
+ wWriteMode.setItems(ParquetWriteMode.getDescriptions());
+ wWriteMode.setToolTipText(BaseMessages.getString(PKG,
"ParquetOutputDialog.WriteMode.ToolTip"));
+ PropsUi.setLook(wWriteMode);
+ FormData fdWriteMode = new FormData();
+ fdWriteMode.left = new FormAttachment(middle, 0);
+ fdWriteMode.top = new FormAttachment(wlWriteMode, 0, SWT.CENTER);
+ fdWriteMode.right = new FormAttachment(100, 0);
+ wWriteMode.setLayoutData(fdWriteMode);
+ lastControl = wWriteMode;
+
+ wlMaxOpenPartitions = new Label(shell, SWT.RIGHT);
+ wlMaxOpenPartitions.setText(
+ BaseMessages.getString(PKG,
"ParquetOutputDialog.MaxOpenPartitions.Label"));
+ wlMaxOpenPartitions.setToolTipText(
+ BaseMessages.getString(PKG,
"ParquetOutputDialog.MaxOpenPartitions.ToolTip"));
+ PropsUi.setLook(wlMaxOpenPartitions);
+ FormData fdlMaxOpenPartitions = new FormData();
+ fdlMaxOpenPartitions.left = new FormAttachment(0, 0);
+ fdlMaxOpenPartitions.right = new FormAttachment(middle, -margin);
+ fdlMaxOpenPartitions.top = new FormAttachment(lastControl, margin);
+ wlMaxOpenPartitions.setLayoutData(fdlMaxOpenPartitions);
+ wMaxOpenPartitions = new TextVar(variables, shell, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
+ wMaxOpenPartitions.enableExpandedInteger();
+ wMaxOpenPartitions.setToolTipText(
+ BaseMessages.getString(PKG,
"ParquetOutputDialog.MaxOpenPartitions.ToolTip"));
+ PropsUi.setLook(wMaxOpenPartitions);
+ FormData fdMaxOpenPartitions = new FormData();
+ fdMaxOpenPartitions.left = new FormAttachment(middle, 0);
+ fdMaxOpenPartitions.top = new FormAttachment(wlMaxOpenPartitions, 0,
SWT.CENTER);
+ fdMaxOpenPartitions.right = new FormAttachment(100, 0);
+ wMaxOpenPartitions.setLayoutData(fdMaxOpenPartitions);
+ lastControl = wMaxOpenPartitions;
+
Label wlFields = new Label(shell, SWT.LEFT);
wlFields.setText(BaseMessages.getString(PKG,
"ParquetOutputDialog.Fields.Label"));
PropsUi.setLook(wlFields);
@@ -468,6 +554,13 @@ public class ParquetOutputDialog extends
BaseTransformDialog {
wlFilenameSplitSize.setEnabled(wFilenameIncludeSplitNr.getSelection());
wFilenameSplitSize.setEnabled(wFilenameIncludeSplitNr.getSelection());
+
+ // The write mode and the open-partition limit only mean something while
partitioning.
+ boolean partitioning = !wPartitionFields.getNonEmptyItems().isEmpty();
+ wlWriteMode.setEnabled(partitioning);
+ wWriteMode.setEnabled(partitioning);
+ wlMaxOpenPartitions.setEnabled(partitioning);
+ wMaxOpenPartitions.setEnabled(partitioning);
}
private void getFields() {
@@ -486,6 +579,7 @@ public class ParquetOutputDialog extends
BaseTransformDialog {
try {
IRowMeta fields = pipelineMeta.getPrevTransformFields(variables,
transformName);
wFields.getColumns()[0].setComboValues(fields.getFieldNames());
+ wPartitionFields.getColumns()[0].setComboValues(fields.getFieldNames());
} catch (Exception e) {
LogChannel.UI.logError("Error getting source fields", e);
}
@@ -512,6 +606,19 @@ public class ParquetOutputDialog extends
BaseTransformDialog {
item.setText(2, Const.NVL(field.getTargetFieldName(), ""));
}
wFields.optimizeTableView();
+
+ for (int i = 0; i < input.getPartitionFields().size(); i++) {
+ ParquetPartitionField field = input.getPartitionFields().get(i);
+ TableItem item = wPartitionFields.table.getItem(i);
+ item.setText(1, Const.NVL(field.getName(), ""));
+ }
+ wPartitionFields.optimizeTableView();
+ wWriteMode.setText(
+ input.getWriteMode() == null
+ ? ParquetWriteMode.Append.getDescription()
+ : input.getWriteMode().getDescription());
+ wMaxOpenPartitions.setText(Const.NVL(input.getMaxOpenPartitions(), ""));
+
enableFields();
}
@@ -546,6 +653,12 @@ public class ParquetOutputDialog extends
BaseTransformDialog {
for (TableItem item : wFields.getNonEmptyItems()) {
input.getFields().add(new ParquetField(item.getText(1),
item.getText(2)));
}
+ input.getPartitionFields().clear();
+ for (TableItem item : wPartitionFields.getNonEmptyItems()) {
+ input.getPartitionFields().add(new
ParquetPartitionField(item.getText(1)));
+ }
+
input.setWriteMode(ParquetWriteMode.getModeFromDescription(wWriteMode.getText()));
+ input.setMaxOpenPartitions(wMaxOpenPartitions.getText());
input.setChanged();
dispose();
}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputMeta.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputMeta.java
index 2cfb84df76..a523a40931 100644
---
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputMeta.java
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetOutputMeta.java
@@ -92,6 +92,15 @@ public class ParquetOutputMeta extends
BaseTransformMeta<ParquetOutput, ParquetO
@HopMetadataProperty(groupKey = "fields", key = "field")
private List<ParquetField> fields;
+ @HopMetadataProperty(groupKey = "partition_fields", key = "partition_field")
+ private List<ParquetPartitionField> partitionFields;
+
+ @HopMetadataProperty(key = "write_mode", storeWithCode = true)
+ private ParquetWriteMode writeMode;
+
+ @HopMetadataProperty(key = "max_open_partitions")
+ private String maxOpenPartitions;
+
public ParquetOutputMeta() {
filenameDateTimeFormat = "yyyyMMdd-HHmmss";
compressionCodec = CompressionCodecName.UNCOMPRESSED;
@@ -100,6 +109,9 @@ public class ParquetOutputMeta extends
BaseTransformMeta<ParquetOutput, ParquetO
dataPageSize = Integer.toString(8192);
dictionaryPageSize =
Integer.toString(ParquetProperties.DEFAULT_DICTIONARY_PAGE_SIZE);
fields = new ArrayList<>();
+ partitionFields = new ArrayList<>();
+ writeMode = ParquetWriteMode.Append;
+ maxOpenPartitions = "10";
filenameIncludingCopyNr = true;
filenameIncludingSplitNr = true;
filenameCreatingParentFolders = true;
@@ -125,6 +137,14 @@ public class ParquetOutputMeta extends
BaseTransformMeta<ParquetOutput, ParquetO
this.dataPageSize = m.dataPageSize;
this.dictionaryPageSize = m.dictionaryPageSize;
this.fields = m.fields;
+ this.partitionFields = new ArrayList<>();
+ if (m.partitionFields != null) {
+ for (ParquetPartitionField f : m.partitionFields) {
+ this.partitionFields.add(new ParquetPartitionField(f));
+ }
+ }
+ this.writeMode = m.writeMode;
+ this.maxOpenPartitions = m.maxOpenPartitions;
}
/**
@@ -398,4 +418,70 @@ public class ParquetOutputMeta extends
BaseTransformMeta<ParquetOutput, ParquetO
public void setFields(List<ParquetField> fields) {
this.fields = fields;
}
+
+ /**
+ * Gets partitionFields
+ *
+ * @return value of partitionFields
+ */
+ public List<ParquetPartitionField> getPartitionFields() {
+ return partitionFields;
+ }
+
+ /**
+ * @param partitionFields The partitionFields to set
+ */
+ public void setPartitionFields(List<ParquetPartitionField> partitionFields) {
+ this.partitionFields = partitionFields;
+ }
+
+ /**
+ * Whether the transform partitions its output. Only then do the write mode
and the maximum number
+ * of open partitions have any effect.
+ *
+ * @return true if at least one partition field is configured
+ */
+ public boolean isPartitioning() {
+ if (partitionFields == null) {
+ return false;
+ }
+ for (ParquetPartitionField field : partitionFields) {
+ if (field.getName() != null && !field.getName().trim().isEmpty()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Gets writeMode
+ *
+ * @return value of writeMode
+ */
+ public ParquetWriteMode getWriteMode() {
+ return writeMode;
+ }
+
+ /**
+ * @param writeMode The writeMode to set
+ */
+ public void setWriteMode(ParquetWriteMode writeMode) {
+ this.writeMode = writeMode;
+ }
+
+ /**
+ * Gets maxOpenPartitions
+ *
+ * @return value of maxOpenPartitions
+ */
+ public String getMaxOpenPartitions() {
+ return maxOpenPartitions;
+ }
+
+ /**
+ * @param maxOpenPartitions The maxOpenPartitions to set
+ */
+ public void setMaxOpenPartitions(String maxOpenPartitions) {
+ this.maxOpenPartitions = maxOpenPartitions;
+ }
}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetPartitionField.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetPartitionField.java
new file mode 100644
index 0000000000..687f3b4634
--- /dev/null
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetPartitionField.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hop.parquet.transforms.output;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+
+/**
+ * An incoming field to partition the output by. The field's value becomes a
{@code name=value}
+ * directory level in the Hive-style layout, and the field itself is not
written into the Parquet
+ * files.
+ */
+@Getter
+@Setter
+public class ParquetPartitionField {
+ @HopMetadataProperty(key = "name")
+ private String name;
+
+ public ParquetPartitionField() {}
+
+ public ParquetPartitionField(String name) {
+ this.name = name;
+ }
+
+ public ParquetPartitionField(ParquetPartitionField f) {
+ this(f.name);
+ }
+}
diff --git
a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteMode.java
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteMode.java
new file mode 100644
index 0000000000..bb7a45aacd
--- /dev/null
+++
b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteMode.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hop.parquet.transforms.output;
+
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.metadata.api.IEnumHasCode;
+
+/**
+ * What to do with data that is already present in a partition folder. Only
applies when the
+ * transform is partitioning its output; without partition fields the
transform writes a single file
+ * as before.
+ */
+public enum ParquetWriteMode implements IEnumHasCode {
+ /** Leave existing files in place and add new ones. This is the
pre-partitioning behaviour. */
+ Append("append", "ParquetOutput.WriteMode.Append"),
+
+ /** Empty a partition folder before the first file of this run is written
into it. */
+ OverwritePartitions("overwrite_partitions",
"ParquetOutput.WriteMode.OverwritePartitions"),
+
+ /** Fail if a partition folder already exists. */
+ FailIfExists("fail_if_exists", "ParquetOutput.WriteMode.FailIfExists"),
+
+ /** Empty the whole base folder once, before the first file of this run is
written. */
+ OverwriteAll("overwrite_all", "ParquetOutput.WriteMode.OverwriteAll");
+
+ private static final Class<?> PKG = ParquetOutputMeta.class;
+
+ private final String code;
+ private final String descriptionKey;
+
+ ParquetWriteMode(String code, String descriptionKey) {
+ this.code = code;
+ this.descriptionKey = descriptionKey;
+ }
+
+ public static String[] getDescriptions() {
+ String[] descriptions = new String[values().length];
+ for (int i = 0; i < descriptions.length; i++) {
+ descriptions[i] = values()[i].getDescription();
+ }
+ return descriptions;
+ }
+
+ public static ParquetWriteMode getModeFromDescription(String description) {
+ for (ParquetWriteMode mode : values()) {
+ if (mode.getDescription().equals(description)) {
+ return mode;
+ }
+ }
+ return Append;
+ }
+
+ @Override
+ public String getCode() {
+ return code;
+ }
+
+ public String getDescription() {
+ return BaseMessages.getString(PKG, descriptionKey);
+ }
+}
diff --git
a/plugins/tech/parquet/src/main/resources/org/apache/hop/parquet/transforms/output/messages/messages_en_US.properties
b/plugins/tech/parquet/src/main/resources/org/apache/hop/parquet/transforms/output/messages/messages_en_US.properties
index db680bb9cb..2f568cf488 100644
---
a/plugins/tech/parquet/src/main/resources/org/apache/hop/parquet/transforms/output/messages/messages_en_US.properties
+++
b/plugins/tech/parquet/src/main/resources/org/apache/hop/parquet/transforms/output/messages/messages_en_US.properties
@@ -17,6 +17,10 @@
ParquetOutput.Description=Writes rows of data to a Parquet file
ParquetOutput.Name=Parquet file output
+ParquetOutput.WriteMode.Append=Append
+ParquetOutput.WriteMode.FailIfExists=Fail if exists
+ParquetOutput.WriteMode.OverwriteAll=Overwrite all
+ParquetOutput.WriteMode.OverwritePartitions=Overwrite partitions
ParquetOutputDialog.CompressionCodec.Label=Compression codec
ParquetOutputDialog.DataPageSize.Label=Data page size
ParquetOutputDialog.DictionaryPageSize.Label=Dictionary page size
@@ -24,8 +28,8 @@ ParquetOutputDialog.Fields.Label=Fields (leave empty to
output all input fields)
ParquetOutputDialog.FieldsColumn.SourceField.Label=Source field
ParquetOutputDialog.FieldsColumn.TargetField.Label=Target field
ParquetOutputDialog.FilenameBase.Label=Base file name
-ParquetOutputDialog.FilenameCreateFolders.Label=Create parent folders
ParquetOutputDialog.FilenameCompressionBeforeExtension.Label=Include
compression codec before extension
+ParquetOutputDialog.FilenameCreateFolders.Label=Create parent folders
ParquetOutputDialog.FilenameDateTimeFormat.Label=Date time format
ParquetOutputDialog.FilenameExtension.Label=Extension
ParquetOutputDialog.FilenameGroup.Label=Filename
@@ -35,8 +39,15 @@ ParquetOutputDialog.FilenameIncludeDateTime.Label=Include
date-time format
ParquetOutputDialog.FilenameIncludeSplitNr.Label=Split into parts and include
number
ParquetOutputDialog.FilenameIncludeTime.Label=Include time
ParquetOutputDialog.FilenameSplitSize.Label=Split size
-ParquetOutputDialog.RowGroupSize.Label Row group size
+ParquetOutputDialog.MaxOpenPartitions.Label=Maximum open partitions
+ParquetOutputDialog.MaxOpenPartitions.ToolTip=How many partition files may be
written at the same time. Each open file buffers up to one row group in memory,
so a lower number uses less memory but writes more files.
+ParquetOutputDialog.PartitionFields.Label=Partition by (leave empty to write a
single file set)
+ParquetOutputDialog.PartitionFieldsColumn.Field.Label=Field
+ParquetOutputDialog.PartitionGroup.Label=Partitioning
+ParquetOutputDialog.RowGroupSize.Label=Row group size
ParquetOutputDialog.TransformName.Label=Transform name
ParquetOutputDialog.Version.Label=Version
-ParquetOutputMeta.keyword=parquet,columnar,write,file,apache
+ParquetOutputDialog.WriteMode.Label=Existing data
+ParquetOutputDialog.WriteMode.ToolTip=What to do with data already present in
a partition folder.
ParquetOutputDialog.extension.Label=Parquet files
+ParquetOutputMeta.keyword=parquet,columnar,write,file,apache
diff --git
a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetOutputPartitionTest.java
b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetOutputPartitionTest.java
new file mode 100644
index 0000000000..efcca79455
--- /dev/null
+++
b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetOutputPartitionTest.java
@@ -0,0 +1,419 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hop.parquet.transforms.output;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Stream;
+import org.apache.hop.core.RowMetaAndData;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Tests for the Hive-style partitioned output of {@link ParquetOutput}. */
+@ExtendWith(RestoreHopEngineEnvironmentExtension.class)
+class ParquetOutputPartitionTest {
+
+ @TempDir private Path tempDir;
+
+ private TransformMockHelper<ParquetOutputMeta, ParquetOutputData> mockHelper;
+
+ @BeforeEach
+ void setUp() {
+ mockHelper =
+ new TransformMockHelper<>(
+ "Parquet Output", ParquetOutputMeta.class,
ParquetOutputData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+ when(mockHelper.pipeline.isRunning()).thenReturn(true);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void partitionPathUsesHiveStyleNameEqualsValue() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region", "year");
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+ output.setInputRowMeta(salesRowMeta());
+ output.resolveOutputFields();
+
+ assertEquals(
+ "region=EU/year=2026",
+ output.partitionPath(salesRowMeta(), new Object[] {1L, "EU", "2026"}));
+ }
+
+ @Test
+ void partitionPathUsesTheHiveDefaultNameForNullAndEmptyValues() throws
Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+ output.setInputRowMeta(salesRowMeta());
+ output.resolveOutputFields();
+
+ assertEquals(
+ "region=" + ParquetOutput.DEFAULT_PARTITION_NAME,
+ output.partitionPath(salesRowMeta(), new Object[] {1L, null, "2026"}));
+ assertEquals(
+ "region=" + ParquetOutput.DEFAULT_PARTITION_NAME,
+ output.partitionPath(salesRowMeta(), new Object[] {1L, "", "2026"}));
+ }
+
+ @Test
+ void pathHostileCharactersAreEscaped() {
+ // A value containing a separator must not create extra folder levels.
+ assertEquals("a%2Fb", ParquetOutput.escapePathValue("a/b"));
+ assertEquals("a%5Cb", ParquetOutput.escapePathValue("a\\b"));
+ assertEquals("a%3Db", ParquetOutput.escapePathValue("a=b"));
+ assertEquals("a%3Ab", ParquetOutput.escapePathValue("a:b"));
+ assertEquals("a%09b", ParquetOutput.escapePathValue("a\tb"));
+ assertEquals("100%25", ParquetOutput.escapePathValue("100%"));
+ assertEquals("plain-value_1",
ParquetOutput.escapePathValue("plain-value_1"));
+ // VFS percent-decodes the path it is handed, so our own escapes are
doubled on the way out
+ assertEquals("region=EU%252FWest",
ParquetOutput.toVfsPath("region=EU%2FWest"));
+ }
+
+ @Test
+ void partitionFolderStripsTrailingSeparators() {
+ ParquetOutputMeta meta = new ParquetOutputMeta();
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+
+ meta.setFilenameBase("/tmp/sales/");
+ assertEquals("/tmp/sales/region=EU", output.partitionFolder("region=EU"));
+
+ meta.setFilenameBase("/tmp/sales\\");
+ assertEquals("/tmp/sales/region=EU", output.partitionFolder("region=EU"));
+
+ meta.setFilenameBase("/tmp/sales");
+ assertEquals("/tmp/sales/region=EU", output.partitionFolder("region=EU"));
+ }
+
+ @Test
+ void partitionFieldsAreNotWrittenIntoTheFile() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ ParquetOutputData data = new ParquetOutputData();
+ ParquetOutput output = createTransform(meta, data);
+ output.setInputRowMeta(salesRowMeta());
+
+ output.resolveOutputFields();
+
+ assertEquals(2, data.outputFields.size());
+ assertEquals("id", data.outputFields.get(0).getSourceFieldName());
+ assertEquals("year", data.outputFields.get(1).getSourceFieldName());
+ assertEquals(List.of(1), data.partitionFieldIndexes);
+ }
+
+ @Test
+ void partitioningOnEveryFieldIsRejected() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("id", "region", "year");
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+ output.setInputRowMeta(salesRowMeta());
+
+ HopException e = assertThrows(HopException.class,
output::resolveOutputFields);
+ assertTrue(e.getMessage().contains("nothing to write"));
+ }
+
+ @Test
+ void unknownPartitionFieldIsRejected() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("nope");
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+ output.setInputRowMeta(salesRowMeta());
+
+ HopException e = assertThrows(HopException.class,
output::resolveOutputFields);
+ assertTrue(e.getMessage().contains("nope"));
+ }
+
+ @Test
+ void duplicatePartitionFieldIsRejected() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region", "region");
+ ParquetOutput output = createTransform(meta, new ParquetOutputData());
+ output.setInputRowMeta(salesRowMeta());
+
+ HopException e = assertThrows(HopException.class,
output::resolveOutputFields);
+ assertTrue(e.getMessage().contains("more than once"));
+ }
+
+ @Test
+ void writesOneFolderPerPartitionAndOmitsThePartitionColumn() throws
Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+
+ runRows(
+ meta,
+ new Object[] {1L, "EU", "2026"},
+ new Object[] {2L, "US", "2026"},
+ new Object[] {3L, "EU", "2026"});
+
+ // Hive-style layout, one folder per distinct value.
+ assertEquals(List.of("region=EU", "region=US"),
names(childFolders(tempDir.resolve("sales"))));
+
+ // The written file no longer carries the partition column ...
+ Path euFile = onlyFile(tempDir.resolve("sales/region=EU"));
+ IRowMeta schema = ParquetTestUtil.readSchema(euFile.toString());
+ assertEquals(2, schema.size());
+ assertEquals("id", schema.getValueMeta(0).getName());
+ assertEquals("year", schema.getValueMeta(1).getName());
+ assertEquals(-1, schema.indexOfValue("region"));
+
+ // ... and both EU rows are in that one file.
+ List<RowMetaAndData> rows = readRows(euFile, "id", "year");
+ assertEquals(2, rows.size());
+ assertEquals(1L, rows.get(0).getInteger("id", -1L));
+ assertEquals(3L, rows.get(1).getInteger("id", -1L));
+ }
+
+ @Test
+ void aPartitionValueWithASeparatorStaysOneFolderLevel() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+
+ runRows(meta, new Object[] {1L, "EU/West", "2026"});
+
+ assertEquals(List.of("region=EU%2FWest"),
names(childFolders(tempDir.resolve("sales"))));
+ }
+
+ @Test
+ void appendModeLeavesEarlierFilesInPlace() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+ meta.setWriteMode(ParquetWriteMode.Append);
+
+ runRows(meta, new Object[] {1L, "EU", "2026"});
+ runRows(meta, new Object[] {2L, "EU", "2026"});
+
+ assertEquals(2, listFiles(tempDir.resolve("sales/region=EU")).size());
+ }
+
+ @Test
+ void overwritePartitionsModeEmptiesOnlyTheTouchedPartition() throws
Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+
+ meta.setWriteMode(ParquetWriteMode.Append);
+ runRows(meta, new Object[] {1L, "EU", "2026"}, new Object[] {2L, "US",
"2026"});
+ assertEquals(1, listFiles(tempDir.resolve("sales/region=EU")).size());
+ assertEquals(1, listFiles(tempDir.resolve("sales/region=US")).size());
+
+ meta.setWriteMode(ParquetWriteMode.OverwritePartitions);
+ runRows(meta, new Object[] {3L, "EU", "2026"});
+
+ // EU was rewritten, US was not touched.
+ assertEquals(1, listFiles(tempDir.resolve("sales/region=EU")).size());
+ assertEquals(1, listFiles(tempDir.resolve("sales/region=US")).size());
+ List<RowMetaAndData> euRows =
readRows(onlyFile(tempDir.resolve("sales/region=EU")), "id");
+ assertEquals(1, euRows.size());
+ assertEquals(3L, euRows.get(0).getInteger("id", -1L));
+ }
+
+ @Test
+ void overwritePartitionsDoesNotDeleteWhatTheSameRunJustWrote() throws
Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+ meta.setWriteMode(ParquetWriteMode.OverwritePartitions);
+ // One open partition at a time forces EU to be closed and reopened
between the two EU rows.
+ meta.setMaxOpenPartitions("1");
+
+ runRows(
+ meta,
+ new Object[] {1L, "EU", "2026"},
+ new Object[] {2L, "US", "2026"},
+ new Object[] {3L, "EU", "2026"});
+
+ // Two files for EU because it was reopened, and the first was not wiped
by the second.
+ assertEquals(2, listFiles(tempDir.resolve("sales/region=EU")).size());
+ }
+
+ @Test
+ void failIfExistsModeRefusesAnExistingPartition() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+
+ meta.setWriteMode(ParquetWriteMode.Append);
+ runRows(meta, new Object[] {1L, "EU", "2026"});
+
+ meta.setWriteMode(ParquetWriteMode.FailIfExists);
+ HopException e =
+ assertThrows(HopException.class, () -> runRows(meta, new Object[] {2L,
"EU", "2026"}));
+ assertTrue(e.getMessage().contains("already exists"));
+ }
+
+ @Test
+ void overwriteAllModeEmptiesEveryPartition() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+
+ meta.setWriteMode(ParquetWriteMode.Append);
+ runRows(meta, new Object[] {1L, "EU", "2026"}, new Object[] {2L, "US",
"2026"});
+
+ meta.setWriteMode(ParquetWriteMode.OverwriteAll);
+ runRows(meta, new Object[] {3L, "APAC", "2026"});
+
+ assertEquals(List.of("region=APAC"),
names(childFolders(tempDir.resolve("sales"))));
+ }
+
+ @Test
+ void moreDistinctValuesThanOpenFilesStillWritesEveryRow() throws Exception {
+ ParquetOutputMeta meta = partitionedMeta("region");
+ meta.setFilenameBase(tempDir.resolve("sales").toString());
+ meta.setMaxOpenPartitions("2");
+
+ List<Object[]> rows = new ArrayList<>();
+ for (int i = 0; i < 6; i++) {
+ rows.add(new Object[] {(long) i, "r" + i, "2026"});
+ }
+ runRows(meta, rows.toArray(new Object[0][]));
+
+ assertEquals(6, childFolders(tempDir.resolve("sales")).size());
+ for (int i = 0; i < 6; i++) {
+ Path folder = tempDir.resolve("sales/region=r" + i);
+ assertEquals(1, listFiles(folder).size(), "expected one file in " +
folder);
+ }
+ }
+
+ @Test
+ void withoutPartitionFieldsNothingChanges() throws Exception {
+ ParquetOutputMeta meta = new ParquetOutputMeta();
+ meta.setFilenameBase(tempDir.resolve("flat").toString());
+ meta.setCompressionCodec(CompressionCodecName.UNCOMPRESSED);
+ meta.setFilenameIncludingCopyNr(false);
+ meta.setFilenameIncludingSplitNr(false);
+ meta.setRowGroupSize("4096");
+ meta.setDataPageSize("1024");
+ meta.setDictionaryPageSize("512");
+
+ assertFalse(meta.isPartitioning());
+ runRows(meta, new Object[] {1L, "EU", "2026"});
+
+ // A single file at the base name, no folders, and the region column is
still written.
+ Path file = tempDir.resolve("flat.parquet");
+ assertTrue(Files.exists(file), "expected " + file);
+ assertEquals(3, ParquetTestUtil.readSchema(file.toString()).size());
+ }
+
+ //
------------------------------------------------------------------------------------------
+
+ private ParquetOutputMeta partitionedMeta(String... partitionFields) {
+ ParquetOutputMeta meta = new ParquetOutputMeta();
+ meta.setCompressionCodec(CompressionCodecName.UNCOMPRESSED);
+ meta.setFilenameIncludingSplitNr(false);
+ // The default row group size is 256MB and every open writer buffers up to
one row group, so
+ // the tests that open several partitions at once would otherwise reserve
gigabytes.
+ meta.setRowGroupSize("4096");
+ meta.setDataPageSize("1024");
+ meta.setDictionaryPageSize("512");
+ for (String field : partitionFields) {
+ meta.getPartitionFields().add(new ParquetPartitionField(field));
+ }
+ return meta;
+ }
+
+ private static IRowMeta salesRowMeta() {
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("id"));
+ rowMeta.addValueMeta(new ValueMetaString("region"));
+ rowMeta.addValueMeta(new ValueMetaString("year"));
+ return rowMeta;
+ }
+
+ /** Feeds the given rows through a fresh transform instance and closes its
files. */
+ private void runRows(ParquetOutputMeta meta, Object[]... rows) throws
Exception {
+ ParquetOutputData data = new ParquetOutputData();
+ ParquetOutput output = spy(createTransform(meta, data));
+ output.setInputRowMeta(salesRowMeta());
+ assertTrue(output.init());
+
+ List<Object[]> remaining = new ArrayList<>(List.of(rows));
+ doNothing().when(output).putRow(any(), any());
+ // doAnswer(...).when(spy) rather than when(spy.getRow()): the latter
invokes the real getRow()
+ // while stubbing, which blocks forever in waitUntilPipelineIsStarted().
+ doAnswer(invocation -> remaining.isEmpty() ? null :
remaining.remove(0)).when(output).getRow();
+
+ while (output.processRow()) {
+ // keep going until the null row closes the files
+ }
+ }
+
+ private ParquetOutput createTransform(ParquetOutputMeta meta,
ParquetOutputData data) {
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ TransformMeta transformMeta = new TransformMeta("Parquet Output", meta);
+ pipelineMeta.addTransform(transformMeta);
+ Pipeline pipeline = new LocalPipelineEngine(pipelineMeta);
+ return new ParquetOutput(transformMeta, meta, data, 0, pipelineMeta,
pipeline);
+ }
+
+ private static List<RowMetaAndData> readRows(Path file, String...
fieldNames) throws Exception {
+ List<org.apache.hop.parquet.transforms.input.ParquetField> fields = new
ArrayList<>();
+ for (String name : fieldNames) {
+ fields.add(
+ new org.apache.hop.parquet.transforms.input.ParquetField(
+ name, name, "id".equals(name) ? "Integer" : "String", null, "0",
"0"));
+ }
+ return ParquetTestUtil.readAllRows(file.toString(), fields);
+ }
+
+ private static List<Path> childFolders(Path parent) throws IOException {
+ try (Stream<Path> stream = Files.list(parent)) {
+ return
stream.filter(Files::isDirectory).sorted(Comparator.naturalOrder()).toList();
+ }
+ }
+
+ private static List<Path> listFiles(Path folder) throws IOException {
+ try (Stream<Path> stream = Files.list(folder)) {
+ return
stream.filter(Files::isRegularFile).sorted(Comparator.naturalOrder()).toList();
+ }
+ }
+
+ private static Path onlyFile(Path folder) throws IOException {
+ List<Path> files = listFiles(folder);
+ assertEquals(1, files.size(), "expected exactly one file in " + folder);
+ return files.get(0);
+ }
+
+ private static List<String> names(List<Path> paths) {
+ return paths.stream().map(p -> p.getFileName().toString()).toList();
+ }
+}
