Issue #3864 : Select logical types on Parquet output fields

docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-output.adoc

@@ -174,13 +174,110 @@ image:tech/parquet/parquet-output-dialog-fields-tab.png[Parquet file output Fiel
 |Fields
 |You can specify which fields to write and in which order.
 You can use the "Get Fields" button to populate the dialog.
-Leave empty to output all input fields.
+Leave the list empty to output all input fields with the types in <<hop-to-parquet-types,Hop to Parquet types>>.
+
+|Source field
+|Incoming field to write.
+
+|Target field
+|Column name in the Parquet file.
+Leave empty to use the source field name.
+
+|Parquet type
+|https://parquet.apache.org/docs/file-format/types/logicaltypes/[Logical or physical type] of the column.
+Leave empty to keep the type chosen from the Hop type.
+Get Fields proposes a type for each field: a Date is proposed as `TimestampMillis`, the same TIMESTAMP(MILLIS) written when the Parquet type is left empty, and a Timestamp is proposed as `TimestampMicros`. Select `Date` for a calendar date, for example so BigQuery loads the column as a DATE.
+Change the proposed type when you need a different one.
+
+|Precision
+|Total number of digits.
+Used only when the Parquet type is `Decimal`.
+Leave empty to use the length of the source field.
+
+|Scale
+|Digits after the decimal point.
+Used only when the Parquet type is `Decimal`.
+Leave empty to use the precision of the source field.
+|===
+
+=== Choosing a Parquet type
 
+[options="header"]
 |===
+|Parquet type|Column|Value
+
+|UTF8
+|STRING
+|Text of the value
+
+|Boolean
+|BOOLEAN
+|Boolean value
+
+|Int32
+|INT32
+|Whole number. A value outside the 32-bit range stops the transform.
+
+|Int64
+|INT64
+|Whole number
+
+|Float
+|FLOAT
+|32-bit floating point number
+
+|Double
+|DOUBLE
+|64-bit floating point number
+
+|Binary
+|BINARY
+|Bytes of the value
+
+|Date
+|DATE
+|Calendar date in the JVM time zone. The time of day is dropped.
+BigQuery and other warehouses load this logical type as a DATE.
+Parquet Input reads it back as midnight of that date in the same time zone.
+
+|TimeMillis
+|TIME(MILLIS), not adjusted to UTC
+|Time of day in the JVM time zone, with millisecond precision
+
+|TimeMicros
+|TIME(MICROS), not adjusted to UTC
+|Time of day in the JVM time zone, with microsecond precision
+
+|TimestampMillis
+|TIMESTAMP(MILLIS), adjusted to UTC
+|Instant, with millisecond precision
+
+|TimestampMicros
+|TIMESTAMP(MICROS), adjusted to UTC
+|Instant, with microsecond precision. The nanoseconds beyond the microsecond are dropped.
+
+|Decimal
+|DECIMAL(precision, scale)
+|Exact decimal, rounded with the rounding type of the field.
+Precision is 1 to 38 and scale is 0 to the precision, taken from the Precision and Scale columns or, when those are empty, from the length and precision of the source field.
+A value with more digits before the decimal point than the column holds stops the transform.
+
+|JSON
+|JSON
+|JSON text
+
+|UUID
+|UUID
+|The 16 bytes of the UUID. The value is the text form, for example `00112233-4455-6677-8899-aabbccddeeff`.
+|===
+
+A field listed without a Parquet type, and every field written because the list is empty, uses the mapping below.
+An existing pipeline that does not set a Parquet type keeps that mapping: a Date stays a TIMESTAMP(MILLIS) until you select `Date`.
 
+[#hop-to-parquet-types]
 == Hop to Parquet types
 
-Every field is written as an optional column of the https://parquet.apache.org/docs/file-format/types/logicaltypes/[Parquet type] below.
+When the Parquet type is left empty, every field is written as an optional column of the https://parquet.apache.org/docs/file-format/types/logicaltypes/[Parquet type] below.
 
 [options="header"]
 |===

plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetField.java

@@ -17,15 +17,37 @@
 
 package org.apache.hop.parquet.transforms.output;
 
+import java.util.Arrays;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.util.Utils;
 import org.apache.hop.metadata.api.HopMetadataProperty;
 
+@Getter
+@Setter
 public class ParquetField {
   @HopMetadataProperty(key = "source_field")
   private String sourceFieldName;
 
   @HopMetadataProperty(key = "target_field")
   private String targetFieldName;
 
+  /**
+   * Parquet type code ({@link ParquetFieldType#getCode()}). Empty keeps the type chosen from the
+   * Hop type of the source field.
+   */
+  @HopMetadataProperty(key = "parquet_type")
+  private String parquetType;
+
+  /** Decimal precision. Used only when the Parquet type is Decimal. */
+  @HopMetadataProperty(key = "precision")
+  private String precision;
+
+  /** Decimal scale. Used only when the Parquet type is Decimal. */
+  @HopMetadataProperty(key = "scale")
+  private String scale;
+
   public ParquetField() {}
 
   public ParquetField(String sourceFieldName, String targetFieldName) {
@@ -34,38 +56,31 @@ public ParquetField(String sourceFieldName, String targetFieldName) {
   }
 
   public ParquetField(ParquetField f) {
-    this(f.sourceFieldName, f.targetFieldName);
+    this.sourceFieldName = f.sourceFieldName;
+    this.targetFieldName = f.targetFieldName;
+    this.parquetType = f.parquetType;
+    this.precision = f.precision;
+    this.scale = f.scale;
   }
 
   /**
-   * Gets sourceFieldName
-   *
-   * @return value of sourceFieldName
+   * @return the selected Parquet type, or null when none is selected
+   * @throws HopException when a type is set but is not one of the supported types
    */
-  public String getSourceFieldName() {
-    return sourceFieldName;
-  }
-
-  /**
-   * @param sourceFieldName The sourceFieldName to set
-   */
-  public void setSourceFieldName(String sourceFieldName) {
-    this.sourceFieldName = sourceFieldName;
-  }
-
-  /**
-   * Gets targetFieldName
-   *
-   * @return value of targetFieldName
-   */
-  public String getTargetFieldName() {
-    return targetFieldName;
-  }
-
-  /**
-   * @param targetFieldName The targetFieldName to set
-   */
-  public void setTargetFieldName(String targetFieldName) {
-    this.targetFieldName = targetFieldName;
+  public ParquetFieldType parquetFieldType() throws HopException {
+    if (Utils.isEmpty(parquetType) || parquetType.trim().isEmpty()) {
+      return null;
+    }
+    ParquetFieldType type = ParquetFieldType.fromCode(parquetType);
+    if (type == null) {
+      throw new HopException(
+          "Parquet type '"
+              + parquetType
+              + "' of field '"
+              + (sourceFieldName == null ? "" : sourceFieldName)
+              + "' is not supported. Supported types: "
+              + String.join(", ", Arrays.asList(ParquetFieldType.codes())));
+    }
+    return type;
   }
 }