This is an automated email from the ASF dual-hosted git repository. hansva pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/hop.git
commit 7a49e2fb380be17839969f8edcaba4ec0e13559e Author: Hans Van Akelyen <[email protected]> AuthorDate: Wed Sep 23 10:18:49 2026 +0200 Hardening parquet input and output, fixes #3598 --- .../pipeline/transforms/parquet-file-input.adoc | 92 +++- .../pipeline/transforms/parquet-file-output.adoc | 51 +- .../parquet/transforms/input/ParquetInputMeta.java | 96 ++-- .../transforms/input/ParquetValueConverter.java | 427 +++++++++------- .../parquet/transforms/output/ParquetOutput.java | 77 ++- .../transforms/output/ParquetWriteSupport.java | 113 ++++- .../input/ParquetInputLogicalTypeTest.java | 212 -------- .../transforms/input/ParquetStreamTest.java | 83 ++++ .../input/ParquetToHopTypeMatrixTest.java | 544 +++++++++++++++++++++ .../input/ParquetValueConverterTest.java | 4 +- .../output/HopToParquetTypeMatrixTest.java | 323 ++++++++++++ .../output/ParquetJsonRoundTripTest.java | 162 ------ .../transforms/output/ParquetWriteSupportTest.java | 86 ++++ 13 files changed, 1641 insertions(+), 629 deletions(-) diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-input.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-input.adoc index 91df004ed9..a5d99f0f8b 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-input.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-input.adoc @@ -32,17 +32,12 @@ For more information on this see: http://parquet.apache.org/[Apache Parquet]. Notes: -* To support reading from any location through Apache VFS each file is loaded into memory (one at a time). -Make sure to allocate enough memory to allow this. -* Long values can be de-serialized to Dates if they are EPOC: milliseconds since `1970-01-01 00:00:00.000` -* Parquet Binary fields are considered to be Hop Strings but you can read them as Hop Binary. -* All input values are passed to the output -* INT96 is converted to the Hop Binary data type. -* Columns annotated as JSON are read as the Hop JSON type, columns annotated as BSON as Hop Binary. -* A column which is stored as an integer in one file and as a floating point number in another can -be read as either Hop Integer or Hop Number: the value is widened or rounded to match the type you -configured for the field. This is useful when reading a set of files whose schemas drifted over -time. +* Files are read in place through Apache VFS: local files directly, other locations (S3, Azure, ...) by seeking in the file. +* All input values are passed to the output. +* A column must have the same type in every file read by the transform, or be read into a Hop type both of its types convert to (see the table below). +A value which can't be converted stops the transform with an error naming the Parquet column, its type and the field. +* Timestamps annotated as adjusted to UTC are instants. +The ones which aren't hold the date and time fields of a local time, taken in the time zone of the Hop JVM. [options="header"] |=== @@ -71,4 +66,77 @@ This row can then be used to extract metadata with the xref:pipeline/transforms/ |Get fields button |With this button you can select a parquet file from which we'll read the schema to populate the Fields grid. -|=== \ No newline at end of file +|=== + +== Parquet to Hop types + +The *Get fields* button proposes a Hop type for every column, based on its https://parquet.apache.org/docs/file-format/types/logicaltypes/[logical type] or, without one, its physical type. +You can pick another Hop type from the last column instead. + +[options="header"] +|=== +|Parquet column|Get fields proposes|Can also be read as + +|BOOLEAN +|Boolean +|String, Integer (1 or 0) + +|INT32, INT64, INT(8/16/32/64, signed), INT(8/16/32, unsigned) +|Integer +|String, Number, BigNumber + +|INT(64, unsigned) +|BigNumber +|String, Number, Integer (up to 9223372036854775807) + +|DECIMAL (on INT32, INT64, FIXED_LEN_BYTE_ARRAY or BINARY) +|BigNumber, with the precision as length and the scale as precision +|String, Number, Integer (the fraction is dropped), Binary (the stored bytes) + +|FLOAT, DOUBLE +|Number +|String, BigNumber, Integer (rounded) + +|FLOAT16 +|Number +|String, BigNumber, Integer (rounded), Binary (the stored bytes) + +|DATE +|Date, midnight in the local time zone +|Timestamp. String, Integer, Number and BigNumber get the stored number of days. + +|TIME (MILLIS, MICROS, NANOS) +|Timestamp, the time on 1970-01-01 +|Date. String, Integer, Number and BigNumber get the stored number. + +|TIMESTAMP (MILLIS, MICROS, NANOS) +|Timestamp, with its full precision +|Date. String, Integer, Number and BigNumber get the stored number. + +|INT96 +|Timestamp +|Date + +|STRING, ENUM +|String +|BigNumber (text holding a number), JSON (text holding JSON), Binary + +|JSON +|JSON +|String, BigNumber (text holding a number), Binary + +|UUID +|UUID, or String when the UUID value type isn't installed +|String, Binary + +|BINARY, FIXED_LEN_BYTE_ARRAY without a logical type +|Binary +|String (UTF-8 text), JSON, BigNumber (text holding a number, or else the unscaled bytes of a decimal with the length and precision of the field) + +|BSON, INTERVAL, GEOMETRY, GEOGRAPHY +|Binary +| + +|LIST, MAP, VARIANT and other nested columns +2+|Not supported +|=== diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-output.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-output.adoc index 3d6b494ba1..1b937e0b9c 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-output.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquet-file-output.adoc @@ -35,9 +35,6 @@ The dialog is split across four tabs so it fits a 1080p screen at 100% zoom: **F Notes: * The date optionally referenced in the output file name(s) will be the start of the pipeline execution. -* Hop Date and Timestamp types are serialized as milliseconds since `1970-01-01 00:00:00.000` UTC, with the Parquet `timestamp-millis` logical type. -* Strings, BigNumbers, JSON and UUID values are written as UTF-8 strings. -JSON columns also carry the Parquet JSON annotation, so that they are read back as JSON rather than as a String. * Rows are buffered in memory until a row group is full (see the *Row group size* option), then encoded, compressed and written. Memory use is therefore bounded by the row group size (times the number of copies and, when partitioning, the number of open partitions), not by the split size. @@ -179,4 +176,50 @@ image:tech/parquet/parquet-output-dialog-fields-tab.png[Parquet file output Fiel You can use the "Get Fields" button to populate the dialog. Leave empty to output all input fields. -|=== \ No newline at end of file +|=== + +== 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. + +[options="header"] +|=== +|Hop type|Parquet column + +|Integer +|INT64 + +|Number +|DOUBLE + +|BigNumber with a length of 1 to 38 +|DECIMAL(length, precision), rounded with the rounding type of the field. +A value with more digits before the decimal point than the column holds stops the transform with an error. + +|BigNumber without a length, or longer than 38 +|STRING, the number as text + +|String +|STRING + +|Boolean +|BOOLEAN + +|Date +|TIMESTAMP(MILLIS), adjusted to UTC + +|Timestamp +|TIMESTAMP(MICROS), adjusted to UTC. +The nanoseconds beyond the microsecond are dropped. + +|Binary +|BINARY + +|JSON +|JSON, the JSON text + +|UUID +|UUID, the 16 bytes of the UUID +|=== + +Other Hop types can't be written to Parquet. 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 b122f437cf..52747de5e2 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 @@ -40,13 +40,16 @@ import org.apache.hop.pipeline.transform.TransformMeta; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.schema.LogicalTypeAnnotation; -import org.apache.parquet.schema.LogicalTypeAnnotation.BsonLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.EnumLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.JsonLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.StringLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.TimeLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; @@ -147,38 +150,7 @@ public class ParquetInputMeta extends BaseTransformMeta<ParquetInput, ParquetInp sourceField += path[i]; } } - PrimitiveType primitiveType = column.getPrimitiveType(); - int hopType = IValueMeta.TYPE_STRING; - LogicalTypeAnnotation logicalType = primitiveType.getLogicalTypeAnnotation(); - if (logicalType != null) { - if ((logicalType instanceof TimestampLogicalTypeAnnotation) - || (logicalType instanceof TimeLogicalTypeAnnotation)) { - hopType = IValueMeta.TYPE_TIMESTAMP; - } else if (logicalType instanceof DateLogicalTypeAnnotation) { - hopType = IValueMeta.TYPE_DATE; - } else if (logicalType instanceof JsonLogicalTypeAnnotation) { - hopType = IValueMeta.TYPE_JSON; - } else if (logicalType instanceof BsonLogicalTypeAnnotation) { - // A BSON document is binary, reading it as text would mangle it. - hopType = IValueMeta.TYPE_BINARY; - } else if (logicalType instanceof DecimalLogicalTypeAnnotation) { - hopType = IValueMeta.TYPE_BIGNUMBER; - } else if (logicalType instanceof IntLogicalTypeAnnotation) { - hopType = IValueMeta.TYPE_INTEGER; - } - } else { - hopType = - switch (primitiveType.getPrimitiveTypeName()) { - case INT32, INT64 -> IValueMeta.TYPE_INTEGER; - case INT96 -> IValueMeta.TYPE_TIMESTAMP; - case FLOAT, DOUBLE -> IValueMeta.TYPE_NUMBER; - case BOOLEAN -> IValueMeta.TYPE_BOOLEAN; - case BINARY -> IValueMeta.TYPE_BINARY; - default -> hopType; - }; - } - IValueMeta valueMeta = ValueMetaFactory.createValueMeta(sourceField, hopType, -1, -1); - rowMeta.addValueMeta(valueMeta); + rowMeta.addValueMeta(hopValueMeta(sourceField, column.getPrimitiveType())); } return rowMeta; } catch (Exception e) { @@ -186,4 +158,62 @@ public class ParquetInputMeta extends BaseTransformMeta<ParquetInput, ParquetInp "Unable to extract row metadata from parquet file '" + filename + "'", e); } } + + /** + * The Hop field a Parquet column is read into by default: from its logical type when it has one + * we know, from its physical type otherwise. + * + * @param name the name of the field + * @param primitiveType the type of the column + * @return the value metadata of the field + * @see <a href="https://parquet.apache.org/docs/file-format/types/logicaltypes/";>Parquet logical + * types</a> + */ + static IValueMeta hopValueMeta(String name, PrimitiveType primitiveType) throws HopException { + LogicalTypeAnnotation logicalType = primitiveType.getLogicalTypeAnnotation(); + int length = -1; + int precision = -1; + int hopType; + if (logicalType instanceof StringLogicalTypeAnnotation + || logicalType instanceof EnumLogicalTypeAnnotation) { + hopType = IValueMeta.TYPE_STRING; + } else if (logicalType instanceof JsonLogicalTypeAnnotation) { + hopType = IValueMeta.TYPE_JSON; + } else if (logicalType instanceof TimestampLogicalTypeAnnotation + || logicalType instanceof TimeLogicalTypeAnnotation) { + hopType = IValueMeta.TYPE_TIMESTAMP; + } else if (logicalType instanceof DateLogicalTypeAnnotation) { + hopType = IValueMeta.TYPE_DATE; + } else if (logicalType instanceof DecimalLogicalTypeAnnotation decimal) { + hopType = IValueMeta.TYPE_BIGNUMBER; + length = decimal.getPrecision(); + precision = decimal.getScale(); + } else if (logicalType instanceof IntLogicalTypeAnnotation intType) { + // An unsigned 64-bit value can be larger than a Hop Integer can hold. + hopType = + !intType.isSigned() && intType.getBitWidth() == 64 + ? IValueMeta.TYPE_BIGNUMBER + : IValueMeta.TYPE_INTEGER; + } else if (logicalType instanceof Float16LogicalTypeAnnotation) { + hopType = IValueMeta.TYPE_NUMBER; + } else if (logicalType instanceof UUIDLogicalTypeAnnotation) { + // The UUID value type is a plugin: fall back to its text when it isn't installed. + hopType = + "-".equals(ValueMetaFactory.getValueMetaName(IValueMeta.TYPE_UUID)) + ? IValueMeta.TYPE_STRING + : IValueMeta.TYPE_UUID; + } else { + // No logical type, or one Hop has no type for (BSON, INTERVAL, GEOMETRY, ...): go by the + // physical type. Binary values we can't interpret are passed on as bytes. + hopType = + switch (primitiveType.getPrimitiveTypeName()) { + case INT32, INT64 -> IValueMeta.TYPE_INTEGER; + case INT96 -> IValueMeta.TYPE_TIMESTAMP; + case FLOAT, DOUBLE -> IValueMeta.TYPE_NUMBER; + case BOOLEAN -> IValueMeta.TYPE_BOOLEAN; + case BINARY, FIXED_LEN_BYTE_ARRAY -> IValueMeta.TYPE_BINARY; + }; + } + return ValueMetaFactory.createValueMeta(name, hopType, length, precision); + } } diff --git a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetValueConverter.java b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetValueConverter.java index c3e2b31761..8e8d0367f3 100644 --- a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetValueConverter.java +++ b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/input/ParquetValueConverter.java @@ -17,17 +17,20 @@ package org.apache.hop.parquet.transforms.input; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import java.math.BigDecimal; import java.math.BigInteger; +import java.math.RoundingMode; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.sql.Timestamp; +import java.time.Instant; import java.time.LocalDate; +import java.time.LocalDateTime; import java.time.ZoneId; +import java.time.ZoneOffset; import java.util.Date; -import java.util.TimeZone; +import java.util.UUID; import org.apache.hop.core.RowMetaAndData; import org.apache.hop.core.exception.HopRuntimeException; import org.apache.hop.core.row.IValueMeta; @@ -36,8 +39,28 @@ import org.apache.parquet.io.api.PrimitiveConverter; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.EnumLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.JsonLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.StringLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimeLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation; import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +/** + * Converts the values of one Parquet column into the Hop type of the field it is mapped to. + * + * <p>The logical type of the column decides what the stored value means (a DECIMAL long is an + * unscaled number, a DATE int a day count, an unsigned int never negative, ...). That meaning is + * worked out first and only then converted to the Hop type the user picked for the field. + * + * @see <a href="https://parquet.apache.org/docs/file-format/types/logicaltypes/";>Parquet logical + * types</a> + */ public class ParquetValueConverter extends PrimitiveConverter { private final RowMetaAndData group; @@ -83,74 +106,86 @@ public class ParquetValueConverter extends PrimitiveConverter { if (rowIndex < 0) { return; } - Object object; - switch (valueMeta.getType()) { - case IValueMeta.TYPE_STRING: - object = value.toStringUsingUTF8(); - break; - case IValueMeta.TYPE_BINARY: - object = value.getBytes(); - break; - case IValueMeta.TYPE_BIGNUMBER: - if (this.logicalTypeAnnotation instanceof DecimalLogicalTypeAnnotation decimal) { - // A DECIMAL column holds the unscaled two's complement value, never text. Trying the - // text route first would misread bytes that happen to be ASCII digits. - object = binaryToDecimal(value, decimal.getPrecision(), decimal.getScale()); - } else { - try { - // Hop itself writes big numbers as strings. - object = new BigDecimal(value.toStringUsingUTF8()); - } catch (NumberFormatException e) { - object = binaryToDecimal(value, valueMeta.getLength(), valueMeta.getPrecision()); - } - } - break; - case IValueMeta.TYPE_JSON: - JsonNode node = null; - try { - ObjectMapper mapper = new ObjectMapper(); - node = mapper.readTree(value.toStringUsingUTF8()); - } catch (Exception e) { - throw new HopRuntimeException("Unable to parse an json value : " + e.getMessage()); - } - object = node; - break; - case IValueMeta.TYPE_TIMESTAMP: - if (value.length() == 12) { - // This is a binary form of an int96 (12-byte) Timestamp with nanosecond precision. - // The first 8 bytes are the nanoseconds in a day. - // The next 4 bytes are the Julian day. - // Note: Little Endian. - // - ByteBuffer bb = ByteBuffer.wrap(value.getBytes()).order(ByteOrder.LITTLE_ENDIAN); - long nsDay = bb.getLong(); - long julianDay = bb.getInt() & 0x00000000ffffffffL; + // Whatever the column holds, a Binary field gets the stored bytes as they are. + // + if (valueMeta.getType() == IValueMeta.TYPE_BINARY) { + set(value.getBytes()); + return; + } + if (logicalTypeAnnotation instanceof DecimalLogicalTypeAnnotation decimal) { + // A DECIMAL column holds the unscaled two's complement value, never text. + setDecimal(binaryToDecimal(value, decimal.getPrecision(), decimal.getScale())); + return; + } + if (logicalTypeAnnotation instanceof Float16LogicalTypeAnnotation) { + // IEEE 754 half precision, little endian. + short bits = ByteBuffer.wrap(value.getBytes()).order(ByteOrder.LITTLE_ENDIAN).getShort(); + addDouble(Float.float16ToFloat(bits)); + return; + } + if (logicalTypeAnnotation instanceof UUIDLogicalTypeAnnotation) { + setUuid(value); + return; + } - // We need a big integer to prevent a long overflow resulting in negative values - // for: nanoseconds since 1970/01/01 00:00:00 - // - BigInteger bns = - BigInteger.valueOf(julianDay - 2440588L) - .multiply(BigInteger.valueOf(86400L * 1000 * 1000 * 1000)) - .add(BigInteger.valueOf(nsDay)); - BigInteger nanosPerSecond = BigInteger.valueOf(1_000_000_000L); - BigInteger[] secondsAndNanos = bns.divideAndRemainder(nanosPerSecond); - BigInteger seconds = secondsAndNanos[0]; - BigInteger nanos = secondsAndNanos[1]; - if (nanos.signum() < 0) { - // Before 1970: keep the nanos positive, as Timestamp requires. - seconds = seconds.subtract(BigInteger.ONE); - nanos = nanos.add(nanosPerSecond); - } - Timestamp timestamp = new Timestamp(seconds.longValue() * 1000L); - timestamp.setNanos(nanos.intValue()); - object = timestamp; - break; - } - default: + boolean int96 = primitiveType.getPrimitiveTypeName() == PrimitiveTypeName.INT96; + if (int96) { + // A legacy timestamp, the only thing it can be read as. + int type = valueMeta.getType(); + if (type != IValueMeta.TYPE_TIMESTAMP && type != IValueMeta.TYPE_DATE) { throw conversionError(); + } + set(int96ToTimestamp(value)); + return; + } + // Only text can be read as text. Without a logical type the bytes are taken to be text too: + // older writers leave strings unannotated. + // + if (!(logicalTypeAnnotation == null + || logicalTypeAnnotation instanceof StringLogicalTypeAnnotation + || logicalTypeAnnotation instanceof EnumLogicalTypeAnnotation + || logicalTypeAnnotation instanceof JsonLogicalTypeAnnotation)) { + throw conversionError(); + } + + Object object = + switch (valueMeta.getType()) { + case IValueMeta.TYPE_STRING -> value.toStringUsingUTF8(); + case IValueMeta.TYPE_BIGNUMBER -> { + try { + // Hop itself writes big numbers without a length as strings. + yield new BigDecimal(value.toStringUsingUTF8()); + } catch (NumberFormatException e) { + if (logicalTypeAnnotation != null) { + throw conversionError(); + } + // Unannotated bytes which aren't a number string: an unscaled decimal, with the + // length and precision of the field standing in for its precision and scale. + yield binaryToDecimal( + value, valueMeta.getLength(), Math.max(valueMeta.getPrecision(), 0)); + } + } + case IValueMeta.TYPE_JSON -> { + try { + yield new ObjectMapper().readTree(value.toStringUsingUTF8()); + } catch (Exception e) { + throw new HopRuntimeException("Unable to parse an json value : " + e.getMessage()); + } + } + default -> throw conversionError(); + }; + set(object); + } + + @Override + public void addInt(int value) { + // An unsigned INT(8/16/32) column uses the full 32 bits, so read it as the unsigned value. + // + if (logicalTypeAnnotation instanceof IntLogicalTypeAnnotation intType && !intType.isSigned()) { + addLong(Integer.toUnsignedLong(value)); + } else { + addLong(value); } - group.getData()[rowIndex] = object; } @Override @@ -158,43 +193,35 @@ public class ParquetValueConverter extends PrimitiveConverter { if (rowIndex < 0) { return; } - Object object; - switch (valueMeta.getType()) { - case IValueMeta.TYPE_INTEGER: - object = value; - break; - case IValueMeta.TYPE_NUMBER: - // An integer column read into a Number field: the same column can be stored as an - // integer in one file and as a double in the next one. - object = (double) value; - break; - case IValueMeta.TYPE_STRING: - object = Long.toString(value); - break; - case IValueMeta.TYPE_DATE: - if (this.logicalTypeAnnotation instanceof DateLogicalTypeAnnotation) { - LocalDate date = LocalDate.ofEpochDay(value); - Date utilDate = Date.from(date.atStartOfDay(ZoneId.systemDefault()).toInstant()); - object = utilDate; - } else { - object = convertToTimestamp(value, this.logicalTypeAnnotation); - } - break; - case IValueMeta.TYPE_BIGNUMBER: - if (this.logicalTypeAnnotation instanceof DecimalLogicalTypeAnnotation decimal) { - // The long is the unscaled value. - object = BigDecimal.valueOf(value, decimal.getScale()); - } else { - object = BigDecimal.valueOf(value); - } - break; - case IValueMeta.TYPE_TIMESTAMP: - object = convertToTimestamp(value, this.logicalTypeAnnotation); - break; - default: - throw conversionError(); + if (logicalTypeAnnotation instanceof DecimalLogicalTypeAnnotation decimal) { + // The long (or int) is the unscaled value. + setDecimal(BigDecimal.valueOf(value, decimal.getScale())); + return; } - group.getData()[rowIndex] = object; + if (logicalTypeAnnotation instanceof IntLogicalTypeAnnotation intType + && !intType.isSigned() + && intType.getBitWidth() == 64) { + // An unsigned 64-bit value doesn't fit in a Java long once it passes Long.MAX_VALUE. + setDecimal(new BigDecimal(Long.toUnsignedString(value))); + return; + } + int type = valueMeta.getType(); + if (type == IValueMeta.TYPE_DATE || type == IValueMeta.TYPE_TIMESTAMP) { + set(toDateOrTimestamp(value, type)); + return; + } + + Object object = + switch (type) { + case IValueMeta.TYPE_INTEGER -> value; + // An integer column read into a Number field: the same column can be stored as an + // integer in one file and as a double in the next one. + case IValueMeta.TYPE_NUMBER -> (double) value; + case IValueMeta.TYPE_STRING -> Long.toString(value); + case IValueMeta.TYPE_BIGNUMBER -> BigDecimal.valueOf(value); + default -> throw conversionError(); + }; + set(object); } @Override @@ -214,7 +241,7 @@ public class ParquetValueConverter extends PrimitiveConverter { case IValueMeta.TYPE_BIGNUMBER -> BigDecimal.valueOf(value); default -> throw conversionError(); }; - group.getData()[rowIndex] = object; + set(object); } @Override @@ -229,7 +256,7 @@ public class ParquetValueConverter extends PrimitiveConverter { case IValueMeta.TYPE_INTEGER -> value ? 1L : 0L; default -> throw conversionError(); }; - group.getData()[rowIndex] = object; + set(object); } @Override @@ -237,93 +264,143 @@ public class ParquetValueConverter extends PrimitiveConverter { addDouble(value); } - @Override - public void addInt(int value) { - addLong(value); + private void set(Object object) { + group.getData()[rowIndex] = object; + } + + /** A DECIMAL or unsigned 64-bit value, into the type of the field. */ + private void setDecimal(BigDecimal decimal) { + Object object = + switch (valueMeta.getType()) { + case IValueMeta.TYPE_BIGNUMBER -> decimal; + case IValueMeta.TYPE_NUMBER -> decimal.doubleValue(); + case IValueMeta.TYPE_STRING -> decimal.toPlainString(); + case IValueMeta.TYPE_INTEGER -> { + try { + // Drop the fraction, as Hop does when converting a BigNumber to an Integer. + yield decimal.setScale(0, RoundingMode.DOWN).longValueExact(); + } catch (ArithmeticException e) { + throw new HopRuntimeException( + "Value " + + decimal.toPlainString() + + " of Parquet column '" + + primitiveType.getName() + + "' doesn't fit in Integer field '" + + valueMeta.getName() + + "'"); + } + } + default -> throw conversionError(); + }; + set(object); + } + + /** A 16-byte big-endian UUID, into the type of the field. */ + private void setUuid(Binary value) { + ByteBuffer buffer = value.toByteBuffer(); + UUID uuid = new UUID(buffer.getLong(), buffer.getLong()); + Object object = + switch (valueMeta.getType()) { + case IValueMeta.TYPE_STRING -> uuid.toString(); + case IValueMeta.TYPE_UUID -> uuid; + default -> throw conversionError(); + }; + set(object); } /** - * Converts a numeric epoch timestamp (in millis, micros, or nanos) into a java.sql.Timestamp - * according to the logical type's time unit. + * An integer column annotated as DATE, TIME or TIMESTAMP, read into a Date or Timestamp field. * - * @param value - * @param logicalTypeAnnotation + * @param value the stored value + * @param type the Hop type of the field: Date or Timestamp + * @return the date or timestamp */ - private Timestamp convertToTimestamp(long value, LogicalTypeAnnotation logicalTypeAnnotation) { - LogicalTypeAnnotation.TimeUnit unit = null; - boolean isUTC = true; - if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) { - unit = - ((LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) logicalTypeAnnotation).getUnit(); - isUTC = - ((LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) this.logicalTypeAnnotation) - .isAdjustedToUTC(); - } else if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.TimeLogicalTypeAnnotation) { - unit = ((LogicalTypeAnnotation.TimeLogicalTypeAnnotation) logicalTypeAnnotation).getUnit(); - isUTC = false; + private Date toDateOrTimestamp(long value, int type) { + if (logicalTypeAnnotation instanceof DateLogicalTypeAnnotation) { + // Days since the epoch: midnight of that day, like any other Hop date without a time. + Date date = + Date.from(LocalDate.ofEpochDay(value).atStartOfDay(ZoneId.systemDefault()).toInstant()); + return type == IValueMeta.TYPE_TIMESTAMP ? new Timestamp(date.getTime()) : date; } - if (unit == null) { - throw new HopRuntimeException( - "Unknown timestamp unit for the logical type: " + logicalTypeAnnotation); + if (logicalTypeAnnotation instanceof TimestampLogicalTypeAnnotation timestamp) { + return toTimestamp(value, timestamp.getUnit(), timestamp.isAdjustedToUTC()); } - long epochMillis = + if (logicalTypeAnnotation instanceof TimeLogicalTypeAnnotation time) { + // A time of day: that time on 1970-01-01. + return toTimestamp(value, time.getUnit(), time.isAdjustedToUTC()); + } + throw conversionError(); + } + + /** + * Converts a number of time units since the epoch into a Timestamp, keeping the sub-millisecond + * part. + * + * @param value the number of units + * @param unit the unit of the value + * @param adjustedToUtc true if the value is an instant, false if it holds the fields of a local + * date and time, which are then taken in the time zone of the JVM + * @return the timestamp + */ + static Timestamp toTimestamp(long value, TimeUnit unit, boolean adjustedToUtc) { + long unitsPerSecond = switch (unit) { - case MILLIS -> value; - case MICROS -> value / 1_000L; - case NANOS -> value / 1_000_000L; - default -> throw new HopRuntimeException("Unknown timestamp unit: " + unit); + case MILLIS -> 1_000L; + case MICROS -> 1_000_000L; + case NANOS -> 1_000_000_000L; }; - // Convert the timestamp to milliseconds since the Epoch based on the original unit - Timestamp ts = new Timestamp(epochMillis); - // If the timestamp is local time, adjust it to UTC - if (!isUTC) { - int offset = TimeZone.getDefault().getOffset(epochMillis); - ts.setTime(ts.getTime() - offset); + // Floor, not truncate: before 1970 the fraction of a second must stay positive. + long seconds = Math.floorDiv(value, unitsPerSecond); + int nanos = (int) (Math.floorMod(value, unitsPerSecond) * (1_000_000_000L / unitsPerSecond)); + if (adjustedToUtc) { + return Timestamp.from(Instant.ofEpochSecond(seconds, nanos)); } - // Adjust nanosecond precision for microsecond or nanosecond timestamps - if (unit == LogicalTypeAnnotation.TimeUnit.MICROS) { - ts.setNanos((int) ((value % 1_000_000L) * 1_000L)); - } else if (unit == LogicalTypeAnnotation.TimeUnit.NANOS) { - ts.setNanos((int) (value % 1_000_000_000L)); + return Timestamp.valueOf(LocalDateTime.ofEpochSecond(seconds, nanos, ZoneOffset.UTC)); + } + + /** + * An INT96 timestamp: nanoseconds in the day (8 bytes) and the Julian day (4 bytes), little + * endian. + */ + private static Timestamp int96ToTimestamp(Binary value) { + ByteBuffer bb = ByteBuffer.wrap(value.getBytes()).order(ByteOrder.LITTLE_ENDIAN); + long nsDay = bb.getLong(); + long julianDay = bb.getInt() & 0x00000000ffffffffL; + + // We need a big integer to prevent a long overflow resulting in negative values + // for: nanoseconds since 1970/01/01 00:00:00 + // + BigInteger bns = + BigInteger.valueOf(julianDay - 2440588L) + .multiply(BigInteger.valueOf(86400L * 1000 * 1000 * 1000)) + .add(BigInteger.valueOf(nsDay)); + BigInteger nanosPerSecond = BigInteger.valueOf(1_000_000_000L); + BigInteger[] secondsAndNanos = bns.divideAndRemainder(nanosPerSecond); + BigInteger seconds = secondsAndNanos[0]; + BigInteger nanos = secondsAndNanos[1]; + if (nanos.signum() < 0) { + // Before 1970: keep the nanos positive, as Timestamp requires. + seconds = seconds.subtract(BigInteger.ONE); + nanos = nanos.add(nanosPerSecond); } - return ts; + Timestamp timestamp = new Timestamp(seconds.longValue() * 1000L); + timestamp.setNanos(nanos.intValue()); + return timestamp; } /** - * Source code from: + * Decodes a DECIMAL stored as bytes: the big-endian two's complement unscaled value. * - * <p>apache/parquet-mr/parquet-pig/src/main/java/org/apache/parquet/pig/convert/DecimalUtils.java - * - * @param value - * @param precision - * @param scale - * @return + * @param value the stored bytes + * @param precision the precision of the decimal, not needed to decode it + * @param scale the number of digits after the decimal point + * @return the exact decimal value */ public static BigDecimal binaryToDecimal(Binary value, int precision, int scale) { - /* - * Precision <= 18 checks for the max number of digits for an unscaled long, - * else treat with big integer conversion - */ - if (precision <= 18) { - ByteBuffer buffer = value.toByteBuffer(); - byte[] bytes = buffer.array(); - int start = buffer.arrayOffset() + buffer.position(); - int end = buffer.arrayOffset() + buffer.limit(); - long unscaled = 0L; - int i = start; - while (i < end) { - unscaled = (unscaled << 8 | bytes[i] & 0xff); - i++; - } - int bits = 8 * (end - start); - long unscaledNew = (unscaled << (64 - bits)) >> (64 - bits); - if (unscaledNew <= -Math.pow(10, 18) || unscaledNew >= Math.pow(10, 18)) { - return new BigDecimal(unscaledNew); - } else { - return BigDecimal.valueOf(unscaledNew / Math.pow(10, scale)); - } - } else { - return new BigDecimal(new BigInteger(value.getBytes()), scale); + byte[] bytes = value.getBytes(); + if (bytes.length == 0) { + return BigDecimal.ZERO.setScale(scale); } + return new BigDecimal(new BigInteger(bytes), scale); } } 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 6811b63cd1..951cc89bfe 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 @@ -23,12 +23,12 @@ import java.text.DecimalFormat; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Locale; import java.util.Map; -import java.util.Set; import java.util.UUID; import org.apache.avro.LogicalTypes; import org.apache.avro.Schema; @@ -55,6 +55,7 @@ import org.apache.parquet.hadoop.ParquetFileWriter; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; import org.apache.parquet.schema.Types; @@ -269,25 +270,47 @@ public class ParquetOutput extends BaseTransform<ParquetOutputMeta, ParquetOutpu } // Convert from Avro to Parquet schema // - return annotateJsonFields(new AvroSchemaConverter().convert(fieldAssembler.endRecord())); + return withParquetOnlyTypes(new AvroSchemaConverter().convert(fieldAssembler.endRecord())); } + /** The largest DECIMAL precision readers such as Spark, Hive and Trino accept. */ + static final int MAX_DECIMAL_PRECISION = 38; + /** - * The Avro type a Hop value is written as. Match these with class ParquetWriteSupport. - * BigDecimal, JSON and UUID values are written as strings, which avoids all sorts of conversion - * issues on the reading side. + * The Avro type a Hop value is written as. Class ParquetWriteSupport writes the values according + * to the Parquet column this becomes. + * + * <ul> + * <li>A Date is an instant with millisecond precision: timestamp-millis. + * <li>A Timestamp keeps its sub-millisecond part: timestamp-micros, the most precise unit + * Spark, Hive and Trino all read. + * <li>A BigNumber with a length (precision) of at most 38 is a DECIMAL with the field's + * precision (scale). Without a length there is no precision to declare, so it is written as + * a string. + * <li>JSON and UUID are strings here: Avro has no JSON type and the Avro converter writes a + * UUID as a string. {@link #withParquetOnlyTypes(MessageType)} gives them their Parquet + * type. + * </ul> */ static Schema avroType(IValueMeta valueMeta) throws HopException { return switch (valueMeta.getType()) { - case IValueMeta.TYPE_TIMESTAMP, IValueMeta.TYPE_DATE -> + case IValueMeta.TYPE_DATE -> LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG)); + case IValueMeta.TYPE_TIMESTAMP -> + LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)); + case IValueMeta.TYPE_BIGNUMBER -> { + int length = valueMeta.getLength(); + int precision = Math.max(valueMeta.getPrecision(), 0); + if (length > 0 && length <= MAX_DECIMAL_PRECISION && precision <= length) { + yield LogicalTypes.decimal(length, precision) + .addToSchema(Schema.create(Schema.Type.BYTES)); + } + yield Schema.create(Schema.Type.STRING); + } case IValueMeta.TYPE_INTEGER -> Schema.create(Schema.Type.LONG); case IValueMeta.TYPE_NUMBER -> Schema.create(Schema.Type.DOUBLE); case IValueMeta.TYPE_BOOLEAN -> Schema.create(Schema.Type.BOOLEAN); - case IValueMeta.TYPE_STRING, - IValueMeta.TYPE_BIGNUMBER, - IValueMeta.TYPE_JSON, - IValueMeta.TYPE_UUID -> + case IValueMeta.TYPE_STRING, IValueMeta.TYPE_JSON, IValueMeta.TYPE_UUID -> Schema.create(Schema.Type.STRING); case IValueMeta.TYPE_BINARY -> Schema.create(Schema.Type.BYTES); default -> @@ -299,32 +322,38 @@ public class ParquetOutput extends BaseTransform<ParquetOutputMeta, ParquetOutpu } /** - * Avro has no JSON logical type, so the schema coming out of the Avro converter describes a Hop - * JSON field as a plain string. Annotate those columns as JSON again so that readers, Parquet - * Input included, recognise them as JSON instead of String. + * Gives the columns Avro can't describe their Parquet type. Avro has no JSON type and the Avro + * converter writes a UUID as a string, so both come out of it as plain strings: + * + * <ul> + * <li>JSON: a string annotated as JSON, so readers, Parquet Input included, see JSON. + * <li>UUID: the 16 bytes of the UUID annotated as UUID. + * </ul> * * @param messageType the schema as converted from Avro - * @return the same schema with the JSON columns annotated + * @return the same schema with the JSON and UUID columns retyped */ - private MessageType annotateJsonFields(MessageType messageType) { - Set<String> jsonFieldNames = new HashSet<>(); + private MessageType withParquetOnlyTypes(MessageType messageType) { + Map<String, Integer> hopTypes = new HashMap<>(); for (int i = 0; i < data.outputFields.size(); i++) { IValueMeta valueMeta = getInputRowMeta().getValueMeta(data.sourceFieldIndexes.get(i)); - if (valueMeta.getType() == IValueMeta.TYPE_JSON) { - jsonFieldNames.add(data.outputFields.get(i).getTargetFieldName()); - } - } - if (jsonFieldNames.isEmpty()) { - return messageType; + hopTypes.put(data.outputFields.get(i).getTargetFieldName(), valueMeta.getType()); } List<Type> types = new ArrayList<>(); for (Type type : messageType.getFields()) { - if (jsonFieldNames.contains(type.getName()) && type.isPrimitive()) { + int hopType = hopTypes.getOrDefault(type.getName(), IValueMeta.TYPE_NONE); + if (hopType == IValueMeta.TYPE_JSON) { types.add( - Types.primitive(type.asPrimitiveType().getPrimitiveTypeName(), type.getRepetition()) + Types.primitive(PrimitiveTypeName.BINARY, type.getRepetition()) .as(LogicalTypeAnnotation.jsonType()) .named(type.getName())); + } else if (hopType == IValueMeta.TYPE_UUID) { + types.add( + Types.primitive(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, type.getRepetition()) + .length(16) + .as(LogicalTypeAnnotation.uuidType()) + .named(type.getName())); } else { types.add(type); } diff --git a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupport.java b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupport.java index 5c1219b28d..b5540718f4 100644 --- a/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupport.java +++ b/plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupport.java @@ -17,16 +17,28 @@ package org.apache.hop.parquet.transforms.output; +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.nio.ByteBuffer; +import java.sql.Timestamp; import java.util.HashMap; import java.util.List; +import java.util.Locale; +import java.util.UUID; import org.apache.hadoop.conf.Configuration; import org.apache.hop.core.RowMetaAndData; import org.apache.hop.core.exception.HopException; import org.apache.hop.core.exception.HopRuntimeException; import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.value.ValueMetaTimestamp; import org.apache.parquet.hadoop.api.WriteSupport; import org.apache.parquet.io.api.Binary; import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; public class ParquetWriteSupport extends WriteSupport<RowMetaAndData> { @@ -36,11 +48,18 @@ public class ParquetWriteSupport extends WriteSupport<RowMetaAndData> { private final List<Integer> sourceFieldIndexes; private final List<ParquetField> fields; + /** The logical type of the column of every field, which decides how its values are stored. */ + private final LogicalTypeAnnotation[] logicalTypes; + public ParquetWriteSupport( MessageType messageType, List<Integer> sourceFieldIndexes, List<ParquetField> fields) { this.messageType = messageType; this.sourceFieldIndexes = sourceFieldIndexes; this.fields = fields; + this.logicalTypes = new LogicalTypeAnnotation[fields.size()]; + for (int i = 0; i < fields.size() && i < messageType.getFieldCount(); i++) { + logicalTypes[i] = messageType.getType(i).getLogicalTypeAnnotation(); + } } @Override @@ -70,21 +89,39 @@ public class ParquetWriteSupport extends WriteSupport<RowMetaAndData> { if (!isNull) { recordConsumer.startField(field.getTargetFieldName(), i); - // Match these data types with ParquetOutput.avroType(). BigDecimal, JSON, UUID and - // anything else go out as strings. + // The column type, as built by ParquetOutput.avroType(), decides how the value is stored. + // Anything without a column type of its own goes out as a string. // + LogicalTypeAnnotation logicalType = logicalTypes[i]; switch (valueMeta.getType()) { case IValueMeta.TYPE_INTEGER -> recordConsumer.addLong(valueMeta.getInteger(valueData)); case IValueMeta.TYPE_NUMBER -> recordConsumer.addDouble(valueMeta.getNumber(valueData)); case IValueMeta.TYPE_BOOLEAN -> recordConsumer.addBoolean(valueMeta.getBoolean(valueData)); case IValueMeta.TYPE_DATE, IValueMeta.TYPE_TIMESTAMP -> - // Epoch milliseconds, as declared by the timestamp-millis logical type. - // The value meta takes care of lazy (binary string) storage and the date mask. - recordConsumer.addLong(valueMeta.getDate(valueData).getTime()); + recordConsumer.addLong(epochValue(valueMeta, valueData, logicalType)); case IValueMeta.TYPE_BINARY -> recordConsumer.addBinary( Binary.fromConstantByteArray(valueMeta.getBinary(valueData))); + case IValueMeta.TYPE_BIGNUMBER -> { + if (logicalType instanceof DecimalLogicalTypeAnnotation decimal) { + recordConsumer.addBinary( + decimalBytes( + field.getTargetFieldName(), + valueMeta, + valueMeta.getBigNumber(valueData), + decimal)); + } else { + recordConsumer.addBinary(Binary.fromString(valueMeta.getString(valueData))); + } + } + case IValueMeta.TYPE_UUID -> { + if (logicalType instanceof UUIDLogicalTypeAnnotation) { + recordConsumer.addBinary(uuidBytes(valueMeta.getString(valueData))); + } else { + recordConsumer.addBinary(Binary.fromString(valueMeta.getString(valueData))); + } + } default -> recordConsumer.addBinary(Binary.fromString(valueMeta.getString(valueData))); } recordConsumer.endField(field.getTargetFieldName(), i); @@ -95,4 +132,70 @@ public class ParquetWriteSupport extends WriteSupport<RowMetaAndData> { throw new HopRuntimeException("Error writing row to Parquet", e); } } + + /** + * A date or timestamp as the number the column holds: microseconds for a TIMESTAMP(MICROS) + * column, milliseconds otherwise. The value meta takes care of lazy (binary string) storage and + * the date mask. + */ + static long epochValue(IValueMeta valueMeta, Object valueData, LogicalTypeAnnotation logicalType) + throws HopException { + if (logicalType instanceof TimestampLogicalTypeAnnotation timestamp + && timestamp.getUnit() == TimeUnit.MICROS) { + Timestamp ts = + valueMeta instanceof ValueMetaTimestamp timestampMeta + ? timestampMeta.getTimestamp(valueData) + : new Timestamp(valueMeta.getDate(valueData).getTime()); + // getTime() holds the whole milliseconds, getNanos() the complete fraction of the second. + return Math.floorDiv(ts.getTime(), 1000L) * 1_000_000L + ts.getNanos() / 1_000L; + } + return valueMeta.getDate(valueData).getTime(); + } + + /** + * A big number as the unscaled two's complement bytes of a DECIMAL column, rounded to the scale + * of the column the way the field rounds. + */ + static Binary decimalBytes( + String fieldName, + IValueMeta valueMeta, + BigDecimal value, + DecimalLogicalTypeAnnotation decimal) { + BigDecimal scaled = value.setScale(decimal.getScale(), roundingMode(valueMeta)); + if (scaled.precision() - scaled.scale() > decimal.getPrecision() - decimal.getScale()) { + throw new HopRuntimeException( + "Value " + + value.toPlainString() + + " of field '" + + fieldName + + "' doesn't fit in DECIMAL(" + + decimal.getPrecision() + + "," + + decimal.getScale() + + "): increase the length of the field"); + } + return Binary.fromConstantByteArray(scaled.unscaledValue().toByteArray()); + } + + private static RoundingMode roundingMode(IValueMeta valueMeta) { + String roundingType = valueMeta.getRoundingType(); + if (roundingType != null) { + try { + return RoundingMode.valueOf(roundingType.toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + // Not a rounding mode we know: use the default. + } + } + return RoundingMode.HALF_EVEN; + } + + /** A UUID as the 16 big-endian bytes of a UUID column. */ + static Binary uuidBytes(String uuid) { + UUID value = UUID.fromString(uuid); + return Binary.fromConstantByteArray( + ByteBuffer.allocate(16) + .putLong(value.getMostSignificantBits()) + .putLong(value.getLeastSignificantBits()) + .array()); + } } diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetInputLogicalTypeTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetInputLogicalTypeTest.java deleted file mode 100644 index ff576e45e0..0000000000 --- a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetInputLogicalTypeTest.java +++ /dev/null @@ -1,212 +0,0 @@ -/* - * 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.input; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; - -import java.math.BigDecimal; -import java.nio.file.Path; -import java.sql.Timestamp; -import java.time.LocalDate; -import java.time.ZoneId; -import java.util.ArrayList; -import java.util.Date; -import java.util.List; -import org.apache.commons.vfs2.FileObject; -import org.apache.hop.core.HopClientEnvironment; -import org.apache.hop.core.RowMetaAndData; -import org.apache.hop.core.vfs.HopVfs; -import org.apache.parquet.example.data.Group; -import org.apache.parquet.example.data.simple.SimpleGroupFactory; -import org.apache.parquet.hadoop.ParquetReader; -import org.apache.parquet.hadoop.ParquetWriter; -import org.apache.parquet.hadoop.example.ExampleParquetWriter; -import org.apache.parquet.io.LocalOutputFile; -import org.apache.parquet.io.api.Binary; -import org.apache.parquet.schema.LogicalTypeAnnotation; -import org.apache.parquet.schema.MessageType; -import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; -import org.apache.parquet.schema.Types; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; - -/** - * Reads a file holding annotated Parquet columns back through the real read path, to guard the - * logical type handling in {@link ParquetValueConverter}. - */ -class ParquetInputLogicalTypeTest { - - /** 2024-01-31, the day the schema mismatch behind issue #3598 was reported. */ - private static final LocalDate DATE = LocalDate.of(2024, 1, 31); - - private static final long EPOCH_MILLIS = 1706716800123L; - - private static final MessageType SCHEMA = - Types.buildMessage() - .optional(PrimitiveTypeName.INT32) - .as(LogicalTypeAnnotation.dateType()) - .named("date_field") - .optional(PrimitiveTypeName.INT64) - .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) - .named("timestamp_millis_field") - .optional(PrimitiveTypeName.INT64) - .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MICROS)) - .named("timestamp_micros_field") - .optional(PrimitiveTypeName.INT64) - .as(LogicalTypeAnnotation.decimalType(2, 18)) - .named("decimal_field") - .optional(PrimitiveTypeName.BINARY) - .as(LogicalTypeAnnotation.jsonType()) - .named("json_field") - .optional(PrimitiveTypeName.BINARY) - .as(LogicalTypeAnnotation.stringType()) - .named("string_field") - .optional(PrimitiveTypeName.INT64) - .named("long_field") - .optional(PrimitiveTypeName.DOUBLE) - .named("double_field") - .named("LogicalTypes"); - - @BeforeAll - static void setUpBeforeAll() throws Exception { - HopClientEnvironment.init(); - } - - private static ParquetField field(String name, String type) { - return new ParquetField(name, name, type, null, "-1", "-1"); - } - - private static Path writeFile(Path folder) throws Exception { - Path file = folder.resolve("logical-types.parquet"); - SimpleGroupFactory groupFactory = new SimpleGroupFactory(SCHEMA); - - try (ParquetWriter<Group> writer = - ExampleParquetWriter.builder(new LocalOutputFile(file)).withType(SCHEMA).build()) { - writer.write( - groupFactory - .newGroup() - .append("date_field", (int) DATE.toEpochDay()) - .append("timestamp_millis_field", EPOCH_MILLIS) - .append("timestamp_micros_field", EPOCH_MILLIS * 1000L) - .append("decimal_field", 123456L) - .append("json_field", Binary.fromString("{\"a\":1}")) - .append("string_field", Binary.fromString("hop")) - .append("long_field", 9081496L) - .append("double_field", 9081496.6d)); - } - return file; - } - - private static RowMetaAndData read(Path file, List<ParquetField> fields) throws Exception { - FileObject fileObject = HopVfs.getFileObject(file.toString()); - try (ParquetStream stream = new ParquetStream(fileObject, file.toString()); - ParquetReader<RowMetaAndData> reader = - new ParquetReaderBuilder<>(new ParquetReadSupport(fields), stream).build()) { - return reader.read(); - } - } - - /** Every annotated column still has to be converted using its logical type. */ - @Test - void testLogicalTypesAreHonoured(@TempDir Path folder) throws Exception { - Path file = writeFile(folder); - - List<ParquetField> fields = new ArrayList<>(); - fields.add(field("date_field", "Date")); - fields.add(field("timestamp_millis_field", "Timestamp")); - fields.add(field("timestamp_micros_field", "Timestamp")); - fields.add(field("decimal_field", "BigNumber")); - fields.add(field("json_field", "JSON")); - fields.add(field("string_field", "String")); - - RowMetaAndData row = read(file, fields); - assertNotNull(row); - - // The DATE annotation makes this an epoch day rather than a number of millis. - Date expectedDate = Date.from(DATE.atStartOfDay(ZoneId.systemDefault()).toInstant()); - assertEquals(expectedDate, row.getData()[0]); - - // Both timestamps describe the same instant, in a different unit. - assertEquals(new Timestamp(EPOCH_MILLIS), row.getData()[1]); - assertEquals(new Timestamp(EPOCH_MILLIS), row.getData()[2]); - - // The DECIMAL annotation carries the scale: 123456 with scale 2 is 1234.56 - assertEquals(0, new BigDecimal("1234.56").compareTo((BigDecimal) row.getData()[3])); - - assertEquals("{\"a\":1}", row.getData()[4].toString()); - assertEquals("hop", row.getData()[5]); - } - - /** - * A DATE column read without its logical type would come out as an instant near the epoch. This - * pins the annotation actually reaching the converter. - */ - @Test - void testDateColumnIsNotReadAsMillis(@TempDir Path folder) throws Exception { - Path file = writeFile(folder); - - RowMetaAndData row = read(file, List.of(field("date_field", "Date"))); - - assertEquals( - DATE, ((Date) row.getData()[0]).toInstant().atZone(ZoneId.systemDefault()).toLocalDate()); - } - - /** The widening conversions added for issue #3598, through the real read path. */ - @Test - void testNumericTypeMismatchIsWidened(@TempDir Path folder) throws Exception { - Path file = writeFile(folder); - - // An int64 column read into a Number field and a double column read into an Integer field. - RowMetaAndData row = - read(file, List.of(field("long_field", "Number"), field("double_field", "Integer"))); - - assertEquals(9081496.0, (Double) row.getData()[0], 0.0); - assertEquals(9081497L, row.getData()[1]); - } - - /** Columns which aren't requested stay out of the row. */ - @Test - void testUnrequestedColumnsAreNotRead(@TempDir Path folder) throws Exception { - Path file = writeFile(folder); - - RowMetaAndData row = read(file, List.of(field("string_field", "String"))); - - assertEquals(1, row.getRowMeta().size()); - assertEquals("hop", row.getData()[0]); - } - - /** Nulls stay null rather than being converted. */ - @Test - void testNullValues(@TempDir Path folder) throws Exception { - Path file = folder.resolve("nulls.parquet"); - SimpleGroupFactory groupFactory = new SimpleGroupFactory(SCHEMA); - try (ParquetWriter<Group> writer = - ExampleParquetWriter.builder(new LocalOutputFile(file)).withType(SCHEMA).build()) { - writer.write(groupFactory.newGroup().append("string_field", Binary.fromString("hop"))); - } - - RowMetaAndData row = - read(file, List.of(field("date_field", "Date"), field("string_field", "String"))); - - assertNull(row.getData()[0]); - assertEquals("hop", row.getData()[1]); - } -} diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetStreamTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetStreamTest.java index 7e474cd9aa..7234177f4b 100644 --- a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetStreamTest.java +++ b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetStreamTest.java @@ -21,11 +21,22 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; +import java.util.List; import org.apache.commons.vfs2.FileObject; import org.apache.hop.core.HopClientEnvironment; +import org.apache.hop.core.RowMetaAndData; import org.apache.hop.core.vfs.HopVfs; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.io.LocalOutputFile; import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -85,4 +96,76 @@ class ParquetStreamTest { assertEquals(filename, stream.toString()); } } + + /** + * A file which isn't local (S3, Azure, ... here the in-memory ram:// file system) is read through + * VFS, seeking back and forth between the footer and the column chunks. + */ + @Test + void testRemoteParquetFileIsReadThroughVfs(@TempDir Path folder) throws Exception { + MessageType schema = + MessageTypeParser.parseMessageType( + "message m { required int64 id; required binary name (STRING); }"); + Path local = folder.resolve("remote.parquet"); + SimpleGroupFactory groups = new SimpleGroupFactory(schema); + try (ParquetWriter<Group> writer = + ExampleParquetWriter.builder(new LocalOutputFile(local)) + .withType(schema) + .withPageSize(1024) + .build()) { + for (long id = 0; id < 1000; id++) { + writer.write(groups.newGroup().append("id", id).append("name", "row " + id)); + } + } + FileObject remote = HopVfs.getFileObject("ram:///parquet/remote.parquet"); + remote.getParent().createFolder(); + try (OutputStream outputStream = HopVfs.getOutputStream(remote, false)) { + outputStream.write(Files.readAllBytes(local)); + } + + List<ParquetField> fields = + List.of( + new ParquetField("id", "id", "Integer", null, "-1", "-1"), + new ParquetField("name", "name", "String", null, "-1", "-1")); + long count = 0; + try (ParquetStream stream = new ParquetStream(remote, remote.toString()); + ParquetReader<RowMetaAndData> reader = + new ParquetReaderBuilder<>(new ParquetReadSupport(fields), stream).build()) { + assertEquals(Files.size(local), stream.getLength()); + RowMetaAndData row; + while ((row = reader.read()) != null) { + assertEquals(count, row.getData()[0]); + assertEquals("row " + count, row.getData()[1]); + count++; + } + } finally { + remote.delete(); + } + assertEquals(1000, count); + } + + /** Seeking a remote file reopens it and skips to the position asked for. */ + @Test + void testRemoteStreamSeeksBothWays() throws Exception { + FileObject remote = HopVfs.getFileObject("ram:///parquet/seek.bin"); + remote.getParent().createFolder(); + try (OutputStream outputStream = HopVfs.getOutputStream(remote, false)) { + outputStream.write(CONTENT); + } + try (ParquetStream stream = new ParquetStream(remote, remote.toString()); + SeekableInputStream inputStream = stream.newStream()) { + inputStream.seek(7); + assertEquals(7, inputStream.getPos()); + assertEquals('H', inputStream.read()); + assertEquals(8, inputStream.getPos()); + + inputStream.seek(0); + byte[] buffer = new byte[6]; + inputStream.readFully(buffer); + assertEquals("Apache", new String(buffer, StandardCharsets.UTF_8)); + assertEquals(6, inputStream.getPos()); + } finally { + remote.delete(); + } + } } diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetToHopTypeMatrixTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetToHopTypeMatrixTest.java new file mode 100644 index 0000000000..05ab191343 --- /dev/null +++ b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetToHopTypeMatrixTest.java @@ -0,0 +1,544 @@ +/* + * 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.input; + +import static org.apache.parquet.schema.LogicalTypeAnnotation.bsonType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.dateType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.decimalType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.enumType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.float16Type; +import static org.apache.parquet.schema.LogicalTypeAnnotation.intType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.jsonType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.stringType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.timeType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.timestampType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.uuidType; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BOOLEAN; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.DynamicContainer.dynamicContainer; +import static org.junit.jupiter.api.DynamicTest.dynamicTest; + +import com.fasterxml.jackson.databind.JsonNode; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.sql.Timestamp; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TimeZone; +import java.util.UUID; +import java.util.function.Consumer; +import java.util.stream.Stream; +import org.apache.hop.core.HopClientEnvironment; +import org.apache.hop.core.RowMetaAndData; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.RowMeta; +import org.apache.hop.core.row.value.ValueMetaBase; +import org.apache.hop.core.row.value.ValueMetaFactory; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.IntervalLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Type; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DynamicNode; +import org.junit.jupiter.api.TestFactory; + +/** + * What every Parquet column type is read as, for every Hop type a field can have: the matrix of the + * Parquet to Hop conversion. + * + * <p>Each column lists the Hop type Get Fields proposes for it and what it reads as in the Hop + * types that accept it. Every Hop type a column doesn't list must fail with a conversion error, so + * the matrix has no gaps. Values are rendered as their Java class and value; the JVM time zone is + * Europe/Brussels (UTC+1 in January) so that UTC and local time can't be confused. + * + * @see <a href="https://parquet.apache.org/docs/file-format/types/logicaltypes/";>Parquet logical + * types</a> + */ +class ParquetToHopTypeMatrixTest { + + /** The Hop types a field can be read into, as named in the matrix. */ + private static final Map<String, Integer> HOP_TYPES = new LinkedHashMap<>(); + + static { + HOP_TYPES.put("String", IValueMeta.TYPE_STRING); + HOP_TYPES.put("Integer", IValueMeta.TYPE_INTEGER); + HOP_TYPES.put("Number", IValueMeta.TYPE_NUMBER); + HOP_TYPES.put("BigNumber", IValueMeta.TYPE_BIGNUMBER); + HOP_TYPES.put("Date", IValueMeta.TYPE_DATE); + HOP_TYPES.put("Timestamp", IValueMeta.TYPE_TIMESTAMP); + HOP_TYPES.put("Boolean", IValueMeta.TYPE_BOOLEAN); + HOP_TYPES.put("Binary", IValueMeta.TYPE_BINARY); + HOP_TYPES.put("JSON", IValueMeta.TYPE_JSON); + HOP_TYPES.put("UUID", IValueMeta.TYPE_UUID); + } + + private static TimeZone defaultTimeZone; + + @BeforeAll + static void setUpBeforeAll() throws Exception { + HopClientEnvironment.init(); + defaultTimeZone = TimeZone.getDefault(); + TimeZone.setDefault(TimeZone.getTimeZone("Europe/Brussels")); + } + + @AfterAll + static void tearDownAfterAll() { + TimeZone.setDefault(defaultTimeZone); + } + + private static List<Column> matrix() { + return List.of( + column("BOOLEAN true", BOOLEAN, null, c -> c.addBoolean(true)) + .getFields("Boolean") + .reads("String", "String true") + .reads("Integer", "Long 1") + .reads("Boolean", "Boolean true"), + column("INT32 42", INT32, null, c -> c.addInt(42)) + .getFields("Integer") + .reads("String", "String 42") + .reads("Integer", "Long 42") + .reads("Number", "Double 42.0") + .reads("BigNumber", "BigDecimal 42"), + column("INT32 INT(8,signed) -5", INT32, intType(8, true), c -> c.addInt(-5)) + .getFields("Integer") + .reads("String", "String -5") + .reads("Integer", "Long -5") + .reads("Number", "Double -5.0") + .reads("BigNumber", "BigDecimal -5"), + column("INT32 INT(32,unsigned) 4294967295", INT32, intType(32, false), c -> c.addInt(-1)) + .getFields("Integer") + .reads("String", "String 4294967295") + .reads("Integer", "Long 4294967295") + .reads("Number", "Double 4.294967295E9") + .reads("BigNumber", "BigDecimal 4294967295"), + column("INT64 42", INT64, null, c -> c.addLong(42L)) + .getFields("Integer") + .reads("String", "String 42") + .reads("Integer", "Long 42") + .reads("Number", "Double 42.0") + .reads("BigNumber", "BigDecimal 42"), + column( + "INT64 INT(64,unsigned) 18446744073709551615", + INT64, + intType(64, false), + c -> c.addLong(-1L)) + .getFields("BigNumber") + .reads("String", "String 18446744073709551615") + .reads("Number", "Double 1.8446744073709552E19") + .reads("BigNumber", "BigDecimal 18446744073709551615"), + column("INT32 DECIMAL(9,2) 12345.67", INT32, decimalType(2, 9), c -> c.addInt(1234567)) + .getFields("BigNumber") + .reads("String", "String 12345.67") + .reads("Integer", "Long 12345") + .reads("Number", "Double 12345.67") + .reads("BigNumber", "BigDecimal 12345.67"), + column("INT64 DECIMAL(18,2) 1234.56", INT64, decimalType(2, 18), c -> c.addLong(123456L)) + .getFields("BigNumber") + .reads("String", "String 1234.56") + .reads("Integer", "Long 1234") + .reads("Number", "Double 1234.56") + .reads("BigNumber", "BigDecimal 1234.56"), + column( + "FIXED[8] DECIMAL(17,2) 123456789012345.67", + fixed(8), + decimalType(2, 17), + c -> c.addBinary(unscaled("123456789012345.67", 8))) + .getFields("BigNumber") + .reads("String", "String 123456789012345.67") + .reads("Integer", "Long 123456789012345") + .reads("Number", "Double 1.2345678901234567E14") + .reads("BigNumber", "BigDecimal 123456789012345.67") + .reads("Binary", "bytes 002bdc545d6b4b87"), + column( + "BINARY DECIMAL(25,3) -1234567890123456789012.345", + BINARY, + decimalType(3, 25), + c -> c.addBinary(unscaled("-1234567890123456789012.345", 0))) + .getFields("BigNumber") + .reads("String", "String -1234567890123456789012.345") + .reads("Number", "Double -1.2345678901234568E21") + .reads("BigNumber", "BigDecimal -1234567890123456789012.345") + .reads("Binary", "bytes fefa91f0c959bbc21d2087"), + column("FLOAT 1.5", FLOAT, null, c -> c.addFloat(1.5F)) + .getFields("Number") + .reads("String", "String 1.5") + .reads("Integer", "Long 2") + .reads("Number", "Double 1.5") + .reads("BigNumber", "BigDecimal 1.5"), + column("DOUBLE 2.5", DOUBLE, null, c -> c.addDouble(2.5D)) + .getFields("Number") + .reads("String", "String 2.5") + .reads("Integer", "Long 3") + .reads("Number", "Double 2.5") + .reads("BigNumber", "BigDecimal 2.5"), + column("FIXED[2] FLOAT16 1.5", fixed(2), float16Type(), c -> c.addBinary(bytes(0x00, 0x3E))) + .getFields("Number") + .reads("String", "String 1.5") + .reads("Integer", "Long 2") + .reads("Number", "Double 1.5") + .reads("BigNumber", "BigDecimal 1.5") + .reads("Binary", "bytes 003e"), + column("INT32 DATE 2024-01-01", INT32, dateType(), c -> c.addInt(19723)) + .getFields("Date") + .reads("String", "String 19723") + .reads("Integer", "Long 19723") + .reads("Number", "Double 19723.0") + .reads("BigNumber", "BigDecimal 19723") + .reads("Date", "Date 2024-01-01 00:00:00.000") + .reads("Timestamp", "Timestamp 2024-01-01 00:00:00.0"), + column( + "INT32 TIME(MILLIS,UTC) 01:00Z", + INT32, + timeType(true, TimeUnit.MILLIS), + c -> c.addInt(3_600_000)) + .getFields("Timestamp") + .reads("String", "String 3600000") + .reads("Integer", "Long 3600000") + .reads("Number", "Double 3600000.0") + .reads("BigNumber", "BigDecimal 3600000") + .reads("Date", "Timestamp 1970-01-01 02:00:00.0") + .reads("Timestamp", "Timestamp 1970-01-01 02:00:00.0"), + column( + "INT64 TIME(MICROS,local) 01:00", + INT64, + timeType(false, TimeUnit.MICROS), + c -> c.addLong(3_600_000_000L)) + .getFields("Timestamp") + .reads("String", "String 3600000000") + .reads("Integer", "Long 3600000000") + .reads("Number", "Double 3.6E9") + .reads("BigNumber", "BigDecimal 3600000000") + .reads("Date", "Timestamp 1970-01-01 01:00:00.0") + .reads("Timestamp", "Timestamp 1970-01-01 01:00:00.0"), + column( + "INT64 TIMESTAMP(MILLIS,UTC) 2024-01-01T00:00:00.123Z", + INT64, + timestampType(true, TimeUnit.MILLIS), + c -> c.addLong(1_704_067_200_123L)) + .getFields("Timestamp") + .reads("String", "String 1704067200123") + .reads("Integer", "Long 1704067200123") + .reads("Number", "Double 1.704067200123E12") + .reads("BigNumber", "BigDecimal 1704067200123") + .reads("Date", "Timestamp 2024-01-01 01:00:00.123") + .reads("Timestamp", "Timestamp 2024-01-01 01:00:00.123"), + column( + "INT64 TIMESTAMP(MICROS,UTC) 2024-01-01T00:00:00.123456Z", + INT64, + timestampType(true, TimeUnit.MICROS), + c -> c.addLong(1_704_067_200_123_456L)) + .getFields("Timestamp") + .reads("String", "String 1704067200123456") + .reads("Integer", "Long 1704067200123456") + .reads("Number", "Double 1.704067200123456E15") + .reads("BigNumber", "BigDecimal 1704067200123456") + .reads("Date", "Timestamp 2024-01-01 01:00:00.123456") + .reads("Timestamp", "Timestamp 2024-01-01 01:00:00.123456"), + column( + "INT64 TIMESTAMP(NANOS,UTC) 2024-01-01T00:00:00.123456789Z", + INT64, + timestampType(true, TimeUnit.NANOS), + c -> c.addLong(1_704_067_200_123_456_789L)) + .getFields("Timestamp") + .reads("String", "String 1704067200123456789") + .reads("Integer", "Long 1704067200123456789") + .reads("Number", "Double 1.7040672001234568E18") + .reads("BigNumber", "BigDecimal 1704067200123456789") + .reads("Date", "Timestamp 2024-01-01 01:00:00.123456789") + .reads("Timestamp", "Timestamp 2024-01-01 01:00:00.123456789"), + column( + "INT64 TIMESTAMP(MICROS,local) 2024-01-01 00:00:00.123456", + INT64, + timestampType(false, TimeUnit.MICROS), + c -> c.addLong(1_704_067_200_123_456L)) + .getFields("Timestamp") + .reads("String", "String 1704067200123456") + .reads("Integer", "Long 1704067200123456") + .reads("Number", "Double 1.704067200123456E15") + .reads("BigNumber", "BigDecimal 1704067200123456") + .reads("Date", "Timestamp 2024-01-01 00:00:00.123456") + .reads("Timestamp", "Timestamp 2024-01-01 00:00:00.123456"), + column( + "INT64 TIMESTAMP(MICROS,UTC) 1969-12-31T23:59:59.999999Z", + INT64, + timestampType(true, TimeUnit.MICROS), + c -> c.addLong(-1L)) + .getFields("Timestamp") + .reads("String", "String -1") + .reads("Integer", "Long -1") + .reads("Number", "Double -1.0") + .reads("BigNumber", "BigDecimal -1") + .reads("Date", "Timestamp 1970-01-01 00:59:59.999999") + .reads("Timestamp", "Timestamp 1970-01-01 00:59:59.999999"), + column( + "INT96 2024-01-01T01:02:03Z", + INT96, + null, + c -> c.addBinary(int96(2460311, 3_723_000_000_000L))) + .getFields("Timestamp") + .reads("Date", "Timestamp 2024-01-01 02:02:03.0") + .reads("Timestamp", "Timestamp 2024-01-01 02:02:03.0") + .reads("Binary", "bytes 00ae17d462030000978a2500"), + column( + "BINARY STRING héllo", + BINARY, + stringType(), + c -> c.addBinary(Binary.fromString("héllo"))) + .getFields("String") + .reads("String", "String héllo") + .reads("Binary", "bytes 68c3a96c6c6f"), + column("BINARY ENUM RED", BINARY, enumType(), c -> c.addBinary(Binary.fromString("RED"))) + .getFields("String") + .reads("String", "String RED") + .reads("Binary", "bytes 524544"), + column( + "BINARY (no annotation) abc", + BINARY, + null, + c -> c.addBinary(Binary.fromString("abc"))) + .getFields("Binary") + .reads("String", "String abc") + .reads("BigNumber", "BigDecimal 6382179") + .reads("Binary", "bytes 616263"), + column( + "BINARY JSON {\"a\":1}", + BINARY, + jsonType(), + c -> c.addBinary(Binary.fromString("{\"a\":1}"))) + .getFields("JSON") + .reads("String", "String {\"a\":1}") + .reads("Binary", "bytes 7b2261223a317d") + .reads("JSON", "JSON {\"a\":1}"), + column("BINARY BSON", BINARY, bsonType(), c -> c.addBinary(bytes(5, 0, 0, 0, 0))) + .getFields("Binary") + .reads("Binary", "bytes 0500000000"), + column( + "FIXED[16] UUID 00112233-4455-6677-8899-aabbccddeeff", + fixed(16), + uuidType(), + c -> c.addBinary(uuid("00112233-4455-6677-8899-aabbccddeeff"))) + .getFields("String") + .reads("String", "String 00112233-4455-6677-8899-aabbccddeeff") + .reads("Binary", "bytes 00112233445566778899aabbccddeeff") + .reads("UUID", "UUID 00112233-4455-6677-8899-aabbccddeeff"), + column( + "FIXED[4] (no annotation) abcd", + fixed(4), + null, + c -> c.addBinary(Binary.fromString("abcd"))) + .getFields("Binary") + .reads("String", "String abcd") + .reads("BigNumber", "BigDecimal 1633837924") + .reads("Binary", "bytes 61626364"), + column( + "FIXED[12] INTERVAL", + fixed(12), + IntervalLogicalTypeAnnotation.getInstance(), + c -> c.addBinary(bytes(1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0))) + .getFields("Binary") + .reads("Binary", "bytes 010000000200000003000000")); + } + + @TestFactory + Stream<DynamicNode> parquetToHop() { + return matrix().stream() + .map( + column -> { + List<DynamicNode> cells = new ArrayList<>(); + cells.add( + dynamicTest( + "Get Fields -> " + column.getFields, + () -> + assertEquals( + column.getFields, + ParquetInputMeta.hopValueMeta("col", column.type).getTypeDesc()))); + for (String hopType : HOP_TYPES.keySet()) { + String expected = column.reads.get(hopType); + cells.add( + dynamicTest( + "-> " + hopType + ": " + (expected == null ? "conversion error" : expected), + () -> { + if (expected == null) { + assertThrows(HopRuntimeException.class, () -> read(column, hopType)); + } else { + assertEquals(expected, render(read(column, hopType))); + } + })); + } + return dynamicContainer(column.label, cells); + }); + } + + private static Object read(Column column, String hopType) throws Exception { + int type = HOP_TYPES.get(hopType); + // The UUID value type is a plugin which isn't on the class path of this test. + IValueMeta valueMeta = + type == IValueMeta.TYPE_UUID + ? new UuidValueMeta() + : ValueMetaFactory.createValueMeta("field", type); + RowMeta rowMeta = new RowMeta(); + rowMeta.addValueMeta(valueMeta); + RowMetaAndData group = new RowMetaAndData(rowMeta, new Object[1]); + column.value.accept(new ParquetValueConverter(group, 0, column.type)); + return group.getData()[0]; + } + + /** A value as its Java class and value, so the matrix shows the type as well. */ + private static String render(Object value) { + if (value instanceof Timestamp timestamp) { + return "Timestamp " + timestamp; + } + if (value instanceof Date date) { + return "Date " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(date); + } + if (value instanceof byte[] bytes) { + return "bytes " + HexFormat.of().formatHex(bytes); + } + if (value instanceof JsonNode json) { + return "JSON " + json; + } + if (value instanceof BigDecimal decimal) { + return "BigDecimal " + decimal.toPlainString(); + } + return value.getClass().getSimpleName() + " " + value; + } + + private static final class Column { + private final String label; + private final PrimitiveType type; + private final Consumer<ParquetValueConverter> value; + private final Map<String, String> reads = new LinkedHashMap<>(); + private String getFields; + + private Column(String label, PrimitiveType type, Consumer<ParquetValueConverter> value) { + this.label = label; + this.type = type; + this.value = value; + } + + Column getFields(String hopType) { + this.getFields = hopType; + return this; + } + + Column reads(String hopType, String expected) { + if (!HOP_TYPES.containsKey(hopType)) { + throw new IllegalArgumentException("Not a Hop type of the matrix: " + hopType); + } + reads.put(hopType, expected); + return this; + } + } + + private static Column column( + String label, + PrimitiveTypeName physicalType, + LogicalTypeAnnotation logicalType, + Consumer<ParquetValueConverter> value) { + return column(label, physicalType, 0, logicalType, value); + } + + private static Column column( + String label, + Fixed fixed, + LogicalTypeAnnotation logicalType, + Consumer<ParquetValueConverter> value) { + return column(label, FIXED_LEN_BYTE_ARRAY, fixed.length, logicalType, value); + } + + private static Column column( + String label, + PrimitiveTypeName physicalType, + int length, + LogicalTypeAnnotation logicalType, + Consumer<ParquetValueConverter> value) { + PrimitiveType type = + new PrimitiveType(Type.Repetition.OPTIONAL, physicalType, length, "col") + .withLogicalTypeAnnotation(logicalType); + return new Column(label, type, value); + } + + private record Fixed(int length) {} + + private static Fixed fixed(int length) { + return new Fixed(length); + } + + private static Binary bytes(int... values) { + byte[] bytes = new byte[values.length]; + for (int i = 0; i < values.length; i++) { + bytes[i] = (byte) values[i]; + } + return Binary.fromConstantByteArray(bytes); + } + + /** The big-endian two's complement unscaled value of a decimal, sign-extended to a length. */ + private static Binary unscaled(String decimal, int length) { + byte[] unscaled = new BigDecimal(decimal).unscaledValue().toByteArray(); + if (length == 0) { + return Binary.fromConstantByteArray(unscaled); + } + byte[] bytes = new byte[length]; + byte sign = unscaled[0] < 0 ? (byte) -1 : 0; + java.util.Arrays.fill(bytes, sign); + System.arraycopy(unscaled, 0, bytes, length - unscaled.length, unscaled.length); + return Binary.fromConstantByteArray(bytes); + } + + private static Binary uuid(String uuid) { + UUID value = UUID.fromString(uuid); + return Binary.fromConstantByteArray( + ByteBuffer.allocate(16) + .putLong(value.getMostSignificantBits()) + .putLong(value.getLeastSignificantBits()) + .array()); + } + + /** An INT96 timestamp: nanoseconds in the day and the Julian day, little endian. */ + private static Binary int96(long julianDay, long nanosOfDay) { + ByteBuffer buffer = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN); + buffer.putLong(nanosOfDay).putInt((int) julianDay); + return Binary.fromConstantByteArray(buffer.array()); + } + + /** Stands in for the UUID value type plugin. */ + private static final class UuidValueMeta extends ValueMetaBase { + UuidValueMeta() { + super("field", IValueMeta.TYPE_UUID); + } + } +} diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetValueConverterTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetValueConverterTest.java index fc0aa940e9..b727842a36 100644 --- a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetValueConverterTest.java +++ b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/input/ParquetValueConverterTest.java @@ -140,7 +140,7 @@ class ParquetValueConverterTest { ByteBuffer buffer = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN); buffer.putLong(nanosInDay).putInt((int) julianDay); - converter(new ValueMetaTimestamp("ts"), null) + converter(new ValueMetaTimestamp("ts"), PrimitiveTypeName.INT96, null) .addBinary(Binary.fromConstantByteArray(buffer.array())); Timestamp timestamp = (Timestamp) value(); @@ -155,7 +155,7 @@ class ParquetValueConverterTest { ByteBuffer buffer = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN); buffer.putLong(86_399_500_000_000L).putInt(2440587); - converter(new ValueMetaTimestamp("ts"), null) + converter(new ValueMetaTimestamp("ts"), PrimitiveTypeName.INT96, null) .addBinary(Binary.fromConstantByteArray(buffer.array())); Timestamp timestamp = (Timestamp) value(); diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/HopToParquetTypeMatrixTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/HopToParquetTypeMatrixTest.java new file mode 100644 index 0000000000..def4327d74 --- /dev/null +++ b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/HopToParquetTypeMatrixTest.java @@ -0,0 +1,323 @@ +/* + * 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.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.DynamicContainer.dynamicContainer; +import static org.junit.jupiter.api.DynamicTest.dynamicTest; +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 com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.math.BigDecimal; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Timestamp; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.HexFormat; +import java.util.List; +import java.util.TimeZone; +import java.util.UUID; +import java.util.stream.Stream; +import org.apache.hop.core.RowMetaAndData; +import org.apache.hop.core.logging.ILoggingObject; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.RowMeta; +import org.apache.hop.core.row.value.ValueMetaBase; +import org.apache.hop.core.row.value.ValueMetaBigNumber; +import org.apache.hop.core.row.value.ValueMetaBinary; +import org.apache.hop.core.row.value.ValueMetaBoolean; +import org.apache.hop.core.row.value.ValueMetaDate; +import org.apache.hop.core.row.value.ValueMetaInteger; +import org.apache.hop.core.row.value.ValueMetaJson; +import org.apache.hop.core.row.value.ValueMetaNumber; +import org.apache.hop.core.row.value.ValueMetaString; +import org.apache.hop.core.row.value.ValueMetaTimestamp; +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.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.LocalInputFile; +import org.apache.parquet.schema.MessageType; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DynamicNode; +import org.junit.jupiter.api.TestFactory; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +/** + * What every Hop type is written as, and what it reads back as: the matrix of the Hop to Parquet + * conversion. + * + * <p>One row holding a value of every Hop type goes through Parquet Output. For every field the + * matrix shows the Parquet column it became, the Hop type Get Fields proposes for that column and + * the value read back into that type. A second row of nulls must read back as nulls. The JVM time + * zone is Europe/Brussels (UTC+1 in January) so that UTC and local time can't be confused. + * + * @see <a href="https://parquet.apache.org/docs/file-format/types/logicaltypes/";>Parquet logical + * types</a> + */ +@ExtendWith(RestoreHopEngineEnvironmentExtension.class) +class HopToParquetTypeMatrixTest { + + private static TimeZone defaultTimeZone; + + @TempDir private static Path tempDir; + + private static MessageType schema; + private static IRowMeta getFields; + private static List<RowMetaAndData> rows; + + private record Field( + IValueMeta valueMeta, + Object value, + String parquetColumn, + String getFieldsType, + String readsBack) {} + + private static List<Field> matrix() throws Exception { + return List.of( + new Field( + new ValueMetaInteger("integer"), 42L, "optional int64 integer", "Integer", "Long 42"), + new Field( + new ValueMetaNumber("number"), 2.5D, "optional double number", "Number", "Double 2.5"), + // Without a length there is no precision to declare: written as a string. + new Field( + bigNumber("bignumber", -1, -1), + new BigDecimal("12345.6789"), + "optional binary bignumber (STRING)", + "String", + "String 12345.6789"), + // With a length: a DECIMAL, rounded half even to the precision (scale) of the field. + new Field( + bigNumber("decimal", 10, 2), + new BigDecimal("1234.565"), + "optional binary decimal (DECIMAL(10,2))", + "BigNumber", + "BigDecimal 1234.56"), + // Longer than the 38 digits Spark, Hive and Trino accept: written as a string. + new Field( + bigNumber("widedecimal", 50, 2), + new BigDecimal("1234.56"), + "optional binary widedecimal (STRING)", + "String", + "String 1234.56"), + new Field( + new ValueMetaString("string"), + "h\u00e9llo", + "optional binary string (STRING)", + "String", + "String h\u00e9llo"), + new Field( + new ValueMetaBoolean("boolean"), + true, + "optional boolean boolean", + "Boolean", + "Boolean true"), + // A Date is an instant with millisecond precision. + new Field( + new ValueMetaDate("date"), + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").parse("2024-01-01 12:34:56.789"), + "optional int64 date (TIMESTAMP(MILLIS,true))", + "Timestamp", + "Timestamp 2024-01-01 12:34:56.789"), + // A Timestamp keeps microseconds: the nanoseconds beyond are dropped. + new Field( + new ValueMetaTimestamp("timestamp"), + Timestamp.valueOf("2024-01-01 12:34:56.123456789"), + "optional int64 timestamp (TIMESTAMP(MICROS,true))", + "Timestamp", + "Timestamp 2024-01-01 12:34:56.123456"), + new Field( + new ValueMetaBinary("binary"), + new byte[] {1, 2, 3}, + "optional binary binary", + "Binary", + "bytes 010203"), + new Field( + new ValueMetaJson("json"), + new ObjectMapper().readTree("{\"a\":1}"), + "optional binary json (JSON)", + "JSON", + "JSON {\"a\":1}"), + // Get Fields proposes a String here because the UUID value type plugin isn't on the class + // path of this test. With the plugin installed it proposes a UUID. + new Field( + new UuidValueMeta("uuid"), + UUID.fromString("00112233-4455-6677-8899-aabbccddeeff"), + "optional fixed_len_byte_array(16) uuid (UUID)", + "String", + "String 00112233-4455-6677-8899-aabbccddeeff")); + } + + @BeforeAll + static void writeAndReadBack() throws Exception { + defaultTimeZone = TimeZone.getDefault(); + TimeZone.setDefault(TimeZone.getTimeZone("Europe/Brussels")); + + List<Field> fields = matrix(); + RowMeta rowMeta = new RowMeta(); + Object[] values = new Object[fields.size()]; + for (int i = 0; i < fields.size(); i++) { + rowMeta.addValueMeta(fields.get(i).valueMeta); + values[i] = fields.get(i).value; + } + Path file = write(rowMeta, values, new Object[fields.size()]); + + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(file))) { + schema = reader.getFooter().getFileMetaData().getSchema(); + } + getFields = ParquetTestUtil.readSchema(file.toString()); + rows = + ParquetTestUtil.readAllRows(file.toString(), ParquetTestUtil.fieldsFromRowMeta(getFields)); + } + + @AfterAll + static void tearDownAfterAll() { + TimeZone.setDefault(defaultTimeZone); + } + + @TestFactory + Stream<DynamicNode> hopToParquet() throws Exception { + return matrix().stream() + .map( + field -> { + String name = field.valueMeta.getName(); + return dynamicContainer( + field.valueMeta.getTypeDesc() + " " + name, + List.of( + dynamicTest( + "written as: " + field.parquetColumn, + () -> assertEquals(field.parquetColumn, schema.getType(name).toString())), + dynamicTest( + "Get Fields -> " + field.getFieldsType, + () -> + assertEquals( + field.getFieldsType, + getFields.searchValueMeta(name).getTypeDesc())), + dynamicTest( + "reads back: " + field.readsBack, + () -> { + RowMetaAndData row = rows.get(0); + assertEquals( + field.readsBack, + render(row.getData()[row.getRowMeta().indexOfValue(name)])); + }), + dynamicTest( + "null reads back as null", + () -> { + RowMetaAndData row = rows.get(1); + assertNull(row.getData()[row.getRowMeta().indexOfValue(name)]); + }))); + }); + } + + private static Path write(IRowMeta rowMeta, Object[]... rowsToWrite) throws Exception { + TransformMockHelper<ParquetOutputMeta, ParquetOutputData> mockHelper = + new TransformMockHelper<>( + "Parquet Output", ParquetOutputMeta.class, ParquetOutputData.class); + try { + when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class))) + .thenReturn(mockHelper.iLogChannel); + when(mockHelper.pipeline.isRunning()).thenReturn(true); + + ParquetOutputMeta meta = new ParquetOutputMeta(); + meta.setCompressionCodec(CompressionCodecName.UNCOMPRESSED); + meta.setFilenameIncludingSplitNr(false); + meta.setFilenameBase(tempDir.resolve("matrix").toString()); + + PipelineMeta pipelineMeta = new PipelineMeta(); + TransformMeta transformMeta = new TransformMeta("Parquet Output", meta); + pipelineMeta.addTransform(transformMeta); + Pipeline pipeline = new LocalPipelineEngine(pipelineMeta); + ParquetOutput output = + spy( + new ParquetOutput( + transformMeta, meta, new ParquetOutputData(), 0, pipelineMeta, pipeline)); + output.setInputRowMeta(rowMeta); + assertTrue(output.init()); + + List<Object[]> remaining = new ArrayList<>(List.of(rowsToWrite)); + doNothing().when(output).putRow(any(), any()); + doAnswer(invocation -> remaining.isEmpty() ? null : remaining.remove(0)) + .when(output) + .getRow(); + while (output.processRow()) { + // keep going until the null row closes the file + } + } finally { + mockHelper.cleanUp(); + } + + try (Stream<Path> files = Files.list(tempDir)) { + return files.filter(Files::isRegularFile).findFirst().orElseThrow(); + } + } + + private static ValueMetaBigNumber bigNumber(String name, int length, int precision) { + ValueMetaBigNumber valueMeta = new ValueMetaBigNumber(name); + valueMeta.setLength(length, precision); + return valueMeta; + } + + /** A value as its Java class and value, so the matrix shows the type as well. */ + private static String render(Object value) { + if (value instanceof Timestamp timestamp) { + return "Timestamp " + timestamp; + } + if (value instanceof Date date) { + return "Date " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(date); + } + if (value instanceof byte[] bytes) { + return "bytes " + HexFormat.of().formatHex(bytes); + } + if (value instanceof JsonNode json) { + return "JSON " + json; + } + if (value instanceof BigDecimal decimal) { + return "BigDecimal " + decimal.toPlainString(); + } + return value.getClass().getSimpleName() + " " + value; + } + + /** Stands in for the UUID value type plugin, which holds java.util.UUID values. */ + private static final class UuidValueMeta extends ValueMetaBase { + UuidValueMeta(String name) { + super(name, IValueMeta.TYPE_UUID); + } + + @Override + public String getString(Object object) { + return object == null ? null : object.toString(); + } + } +} diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetJsonRoundTripTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetJsonRoundTripTest.java deleted file mode 100644 index 85c777aedc..0000000000 --- a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetJsonRoundTripTest.java +++ /dev/null @@ -1,162 +0,0 @@ -/* - * 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.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 com.fasterxml.jackson.databind.ObjectMapper; -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.logging.ILoggingObject; -import org.apache.hop.core.row.IRowMeta; -import org.apache.hop.core.row.IValueMeta; -import org.apache.hop.core.row.RowMeta; -import org.apache.hop.core.row.value.ValueMetaJson; -import org.apache.hop.core.row.value.ValueMetaString; -import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension; -import org.apache.hop.parquet.transforms.input.ParquetField; -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; - -/** - * Avro, which the output schema is built with, has no JSON logical type. These tests pin that a Hop - * JSON field still comes back out of the file as a JSON field rather than as a String. - */ -@ExtendWith(RestoreHopEngineEnvironmentExtension.class) -class ParquetJsonRoundTripTest { - - private static final String JSON = "{\"id\":1,\"tags\":[\"a\",\"b\"]}"; - - @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(); - } - - /** A Hop JSON field has to be annotated as JSON in the schema of the file we write. */ - @Test - void testJsonFieldIsReadBackAsJson() throws Exception { - Path file = writeOneRow(); - - IRowMeta rowMeta = ParquetTestUtil.readSchema(file.toString()); - - assertEquals(IValueMeta.TYPE_JSON, rowMeta.getValueMeta(rowMeta.indexOfValue("doc")).getType()); - // A plain string field has to stay a String, so the annotation isn't applied to everything. - assertEquals( - IValueMeta.TYPE_STRING, rowMeta.getValueMeta(rowMeta.indexOfValue("name")).getType()); - } - - /** And the value itself has to survive the round trip. */ - @Test - void testJsonValueSurvivesTheRoundTrip() throws Exception { - Path file = writeOneRow(); - - List<ParquetField> fields = - List.of( - new ParquetField("doc", "doc", "JSON", null, "-1", "-1"), - new ParquetField("name", "name", "String", null, "-1", "-1")); - List<RowMetaAndData> rows = ParquetTestUtil.readAllRows(file.toString(), fields); - - assertEquals(1, rows.size()); - RowMetaAndData row = rows.get(0); - assertEquals(IValueMeta.TYPE_JSON, row.getValueMeta(0).getType()); - assertEquals( - new ObjectMapper().readTree(JSON), new ObjectMapper().readTree(row.getString(0, ""))); - assertEquals("hop", row.getString(1, "")); - } - - private Path writeOneRow() throws Exception { - ParquetOutputMeta meta = new ParquetOutputMeta(); - meta.setCompressionCodec(CompressionCodecName.UNCOMPRESSED); - meta.setFilenameIncludingSplitNr(false); - meta.setFilenameBase(tempDir.resolve("docs").toString()); - meta.setRowGroupSize("4096"); - meta.setDataPageSize("1024"); - meta.setDictionaryPageSize("512"); - - ParquetOutputData data = new ParquetOutputData(); - PipelineMeta pipelineMeta = new PipelineMeta(); - TransformMeta transformMeta = new TransformMeta("Parquet Output", meta); - pipelineMeta.addTransform(transformMeta); - Pipeline pipeline = new LocalPipelineEngine(pipelineMeta); - ParquetOutput output = - spy(new ParquetOutput(transformMeta, meta, data, 0, pipelineMeta, pipeline)); - - RowMeta rowMeta = new RowMeta(); - rowMeta.addValueMeta(new ValueMetaJson("doc")); - rowMeta.addValueMeta(new ValueMetaString("name")); - output.setInputRowMeta(rowMeta); - assertTrue(output.init()); - - // A single row, spelled out rather than List.of() so it stays a list of one Object[]. - List<Object[]> remaining = new ArrayList<>(); - remaining.add(new Object[] {new ObjectMapper().readTree(JSON), "hop"}); - - doNothing().when(output).putRow(any(), any()); - doAnswer(invocation -> remaining.isEmpty() ? null : remaining.remove(0)).when(output).getRow(); - - while (output.processRow()) { - // keep going until the null row closes the file - } - - return onlyFile(tempDir); - } - - private static Path onlyFile(Path folder) throws IOException { - try (Stream<Path> stream = Files.list(folder)) { - List<Path> files = - stream.filter(Files::isRegularFile).sorted(Comparator.naturalOrder()).toList(); - assertEquals(1, files.size(), () -> "Expected one file in " + folder + " but got " + files); - return files.get(0); - } - } -} diff --git a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupportTest.java b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupportTest.java index 68ee378c85..f8d517bb6d 100644 --- a/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupportTest.java +++ b/plugins/tech/parquet/src/test/java/org/apache/hop/parquet/transforms/output/ParquetWriteSupportTest.java @@ -17,9 +17,14 @@ package org.apache.hop.parquet.transforms.output; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import java.math.BigDecimal; +import java.math.BigInteger; import java.nio.charset.StandardCharsets; import java.sql.Timestamp; import java.util.Date; @@ -29,8 +34,10 @@ import org.apache.avro.Schema; import org.apache.avro.SchemaBuilder; import org.apache.hadoop.conf.Configuration; import org.apache.hop.core.RowMetaAndData; +import org.apache.hop.core.exception.HopRuntimeException; import org.apache.hop.core.row.IValueMeta; import org.apache.hop.core.row.RowMeta; +import org.apache.hop.core.row.value.ValueMetaBigNumber; import org.apache.hop.core.row.value.ValueMetaDate; import org.apache.hop.core.row.value.ValueMetaInteger; import org.apache.hop.core.row.value.ValueMetaString; @@ -38,6 +45,8 @@ import org.apache.hop.core.row.value.ValueMetaTimestamp; import org.apache.parquet.avro.AvroSchemaConverter; import org.apache.parquet.hadoop.api.WriteSupport; import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.junit.jupiter.api.Test; @@ -140,4 +149,81 @@ class ParquetWriteSupportTest { verify(consumer).addLong(2_000L); verify(consumer).addLong(3_000L); } + + private static final DecimalLogicalTypeAnnotation DECIMAL_10_2 = + (DecimalLogicalTypeAnnotation) LogicalTypeAnnotation.decimalType(2, 10); + + private static BigDecimal decimal(String fieldValue, ValueMetaBigNumber valueMeta) { + byte[] bytes = + ParquetWriteSupport.decimalBytes("n", valueMeta, new BigDecimal(fieldValue), DECIMAL_10_2) + .getBytes(); + return new BigDecimal(new BigInteger(bytes), DECIMAL_10_2.getScale()); + } + + @Test + void testDecimalIsRoundedTheWayTheFieldRounds() { + ValueMetaBigNumber halfEven = new ValueMetaBigNumber("n"); + assertEquals(new BigDecimal("1234.56"), decimal("1234.565", halfEven)); + assertEquals(new BigDecimal("1234.58"), decimal("1234.575", halfEven)); + + ValueMetaBigNumber halfUp = new ValueMetaBigNumber("n"); + halfUp.setRoundingType("half_up"); + assertEquals(new BigDecimal("1234.57"), decimal("1234.565", halfUp)); + + ValueMetaBigNumber unknown = new ValueMetaBigNumber("n"); + unknown.setRoundingType("sideways"); + assertEquals(new BigDecimal("1234.56"), decimal("1234.565", unknown)); + } + + @Test + void testDecimalWhichDoesNotFitFailsWithTheFieldName() { + // DECIMAL(10,2) leaves 8 digits before the point. + assertEquals( + new BigDecimal("99999999.99"), decimal("99999999.99", new ValueMetaBigNumber("n"))); + + HopRuntimeException e = + assertThrows( + HopRuntimeException.class, + () -> + ParquetWriteSupport.decimalBytes( + "amount", + new ValueMetaBigNumber("amount"), + new BigDecimal("123456789.00"), + DECIMAL_10_2)); + assertTrue(e.getMessage().contains("'amount'"), e.getMessage()); + assertTrue(e.getMessage().contains("DECIMAL(10,2)"), e.getMessage()); + } + + @Test + void testTimestampMicrosKeepTheFractionAlsoBeforeTheEpoch() throws Exception { + LogicalTypeAnnotation micros = + LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MICROS); + ValueMetaTimestamp valueMeta = new ValueMetaTimestamp("ts"); + + Timestamp after = new Timestamp(1_000L); + after.setNanos(123_456_789); + assertEquals(1_123_456L, ParquetWriteSupport.epochValue(valueMeta, after, micros)); + + // One microsecond before 1970: -1, not -1001 or +999999. + Timestamp before = new Timestamp(-1_000L); + before.setNanos(999_999_000); + assertEquals(-1L, ParquetWriteSupport.epochValue(valueMeta, before, micros)); + + // A Date in a TIMESTAMP(MICROS) column: its milliseconds, in microseconds. + assertEquals( + 2_000_000L, + ParquetWriteSupport.epochValue(new ValueMetaDate("d"), new Date(2_000L), micros)); + + // Any other column: milliseconds. + assertEquals(1_123L, ParquetWriteSupport.epochValue(valueMeta, after, null)); + } + + @Test + void testUuidBytesAreBigEndian() { + byte[] bytes = ParquetWriteSupport.uuidBytes("00112233-4455-6677-8899-aabbccddeeff").getBytes(); + assertEquals(16, bytes.length); + assertEquals(0x00, bytes[0]); + assertEquals(0x11, bytes[1]); + assertEquals((byte) 0xff, bytes[15]); + } }
