hansva commented on code in PR #9211: URL: https://github.com/apache/hop/pull/9211#discussion_r4444179251
########## plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/ReadParquetFileFn.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.beam.core.fn; + +import java.util.Arrays; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.hadoop.conf.Configuration; +import org.apache.hop.beam.core.BeamHop; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.pipeline.Pipeline; +import org.apache.parquet.avro.AvroParquetReader; +import org.apache.parquet.avro.AvroReadSupport; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.io.LocalInputFile; + +/** + * Reads the Parquet file named by one incoming field and appends the configured columns. + * + * <p>{@code ParquetIO.readFiles} returns records without the path, so the source row would be lost. + * This reads the file on the worker that holds the row, with the same Avro projection {@code + * ParquetIO.read(schema)} would use. + */ +@RequiredArgsConstructor +public class ReadParquetFileFn extends DoFn<HopRow, HopRow> { + private final String transformName; + private final String rowMetaJson; + private final String filenameField; + private final String schemaJson; + private final List<ParquetColumnMapping> columns; + + private transient IRowMeta rowMeta; + private transient int filenameIndex; + private transient int inputSize; + private transient Schema schema; + private transient Counter readCounter; + private transient Counter outputCounter; + private transient Counter errorCounter; + + @Setup + public void setup() { + readCounter = Metrics.counter(Pipeline.METRIC_NAME_READ, transformName); + outputCounter = Metrics.counter(Pipeline.METRIC_NAME_OUTPUT, transformName); + errorCounter = Metrics.counter(Pipeline.METRIC_NAME_ERROR, transformName); + try { + BeamHop.init(); + rowMeta = JsonRowMeta.fromJson(rowMetaJson); + filenameIndex = rowMeta.indexOfValue(filenameField); + if (filenameIndex < 0) { + throw new HopRuntimeException("Filename field '" + filenameField + "' was not found"); + } + inputSize = rowMeta.size(); + schema = new Schema.Parser().parse(schemaJson); + Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException("Unable to initialize Parquet input " + transformName, e); + } + } + + @ProcessElement + public void process(@Element HopRow row, OutputReceiver<HopRow> output) { + try { + String filename = rowMeta.getString(row.getRow(), filenameIndex); + if (filename == null || filename.isEmpty()) { + return; + } + Configuration configuration = new Configuration(false); + AvroReadSupport.setRequestedProjection(configuration, schema); + try (ParquetReader<GenericRecord> reader = + AvroParquetReader.<GenericRecord>builder( + new LocalInputFile(java.nio.file.Path.of(filename))) Review Comment: **[bug]** `LocalInputFile(java.nio.file.Path.of(filename))` only opens local paths. On Dataflow, Spark or Flink the files are almost always `gs://`, `s3://` or `hdfs://`, and `Path.of` cannot open them (an invalid path, or a relative file that does not exist on the worker). The Avro input in #9210 goes through Beam `FileSystems` for this reason. **Suggestion:** match and open the file with `FileSystems.matchSingleFileSpec` / `FileSystems.open(...)` and wrap the `SeekableByteChannel` in a Parquet `InputFile`, the way `ParquetIO` does internally for `ReadableFile.openSeekable()`. A test with a `file://` URI would at least cover the scheme handling. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/ReadParquetFileFn.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.beam.core.fn; + +import java.util.Arrays; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.hadoop.conf.Configuration; +import org.apache.hop.beam.core.BeamHop; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.pipeline.Pipeline; +import org.apache.parquet.avro.AvroParquetReader; +import org.apache.parquet.avro.AvroReadSupport; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.io.LocalInputFile; + +/** + * Reads the Parquet file named by one incoming field and appends the configured columns. + * + * <p>{@code ParquetIO.readFiles} returns records without the path, so the source row would be lost. + * This reads the file on the worker that holds the row, with the same Avro projection {@code + * ParquetIO.read(schema)} would use. + */ +@RequiredArgsConstructor +public class ReadParquetFileFn extends DoFn<HopRow, HopRow> { + private final String transformName; + private final String rowMetaJson; + private final String filenameField; + private final String schemaJson; + private final List<ParquetColumnMapping> columns; + + private transient IRowMeta rowMeta; + private transient int filenameIndex; + private transient int inputSize; + private transient Schema schema; + private transient Counter readCounter; + private transient Counter outputCounter; + private transient Counter errorCounter; + + @Setup + public void setup() { + readCounter = Metrics.counter(Pipeline.METRIC_NAME_READ, transformName); + outputCounter = Metrics.counter(Pipeline.METRIC_NAME_OUTPUT, transformName); + errorCounter = Metrics.counter(Pipeline.METRIC_NAME_ERROR, transformName); + try { + BeamHop.init(); + rowMeta = JsonRowMeta.fromJson(rowMetaJson); + filenameIndex = rowMeta.indexOfValue(filenameField); + if (filenameIndex < 0) { + throw new HopRuntimeException("Filename field '" + filenameField + "' was not found"); + } + inputSize = rowMeta.size(); + schema = new Schema.Parser().parse(schemaJson); + Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException("Unable to initialize Parquet input " + transformName, e); + } + } + + @ProcessElement + public void process(@Element HopRow row, OutputReceiver<HopRow> output) { + try { + String filename = rowMeta.getString(row.getRow(), filenameIndex); + if (filename == null || filename.isEmpty()) { + return; + } + Configuration configuration = new Configuration(false); + AvroReadSupport.setRequestedProjection(configuration, schema); + try (ParquetReader<GenericRecord> reader = + AvroParquetReader.<GenericRecord>builder( + new LocalInputFile(java.nio.file.Path.of(filename))) + .withConf(configuration) + .build()) { + GenericRecord record; + while ((record = reader.read()) != null) { + Object[] copy = Arrays.copyOf(row.getRow(), inputSize + columns.size()); + for (int i = 0; i < columns.size(); i++) { + copy[inputSize + i] = + hopValue(record.get(columns.get(i).getColumnName()), columns.get(i)); + } + output.output(new HopRow(copy)); + readCounter.inc(); + outputCounter.inc(); + } + } + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException( + "Unable to read a Parquet file for transform " + transformName, e); + } + } + + private static Object hopValue(Object value, ParquetColumnMapping column) { Review Comment: **[bug]** The projection is the same Avro schema, read with the plain `GenericData` model, so logical types are not converted. `hopValue` only converts String, Integer and Number: - Date and Timestamp columns get a `Long` (millis or micros) instead of a `Date` / `Timestamp`. - BigNumber (decimal) and Binary columns get a `ByteBuffer` instead of a `BigDecimal` / `byte[]`. The next transform then fails with a ClassCastException or works with wrong values. **Suggestion:** convert per Hop type, including the logical type of the column (`timestamp-millis` / `timestamp-micros` / `decimal`), or read with a `GenericData` that has the Avro logical-type conversions registered. A Date, Timestamp, BigNumber and Binary column in the round-trip test would cover it. ########## plugins/engines/beam/pom.xml: ########## @@ -1182,6 +1182,23 @@ <groupId>org.apache.beam</groupId> <artifactId>beam-sdks-java-io-mqtt</artifactId> </dependency> + + <!-- Parquet file input and output. Kept out of lib/core: the Parquet plugin Review Comment: **[nit]** The reason in this comment is not accurate: `beam-sdks-java-io-parquet` and the Parquet plugin both resolve to Parquet 1.18.1 (`dependency:tree` on both modules). Because `dependencies.xml` now also puts `tech/parquet` on the Beam loader, that loader sees two identical copies of the Parquet jars. That is harmless at the same version, but the comment should say what actually holds, so nobody relies on the plugin having a newer parquet-avro. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/HopRowToParquetFn.java: ########## @@ -0,0 +1,126 @@ +/* + * 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.beam.core.fn; + +import java.util.Date; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.hop.beam.core.BeamHop; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.pipeline.Pipeline; + +/** Maps one Hop row onto the flat Avro record ParquetIO writes. */ +@RequiredArgsConstructor +public class HopRowToParquetFn extends DoFn<HopRow, GenericRecord> { + private final String transformName; + private final String rowMetaJson; + private final String schemaJson; + private final List<String> sourceFields; + + private transient Schema schema; + private transient IRowMeta rowMeta; + private transient int[] indexes; + private transient Counter readCounter; + private transient Counter outputCounter; + private transient Counter errorCounter; + + @Setup + public void setup() { + readCounter = Metrics.counter(Pipeline.METRIC_NAME_READ, transformName); + outputCounter = Metrics.counter(Pipeline.METRIC_NAME_OUTPUT, transformName); + errorCounter = Metrics.counter(Pipeline.METRIC_NAME_ERROR, transformName); + try { + BeamHop.init(); + schema = new Schema.Parser().parse(schemaJson); + rowMeta = JsonRowMeta.fromJson(rowMetaJson); + indexes = new int[sourceFields.size()]; + for (int i = 0; i < sourceFields.size(); i++) { + indexes[i] = rowMeta.indexOfValue(sourceFields.get(i)); + if (indexes[i] < 0) { + throw new HopRuntimeException("Field [" + sourceFields.get(i) + "] was not found"); + } + } + Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException("Unable to initialize Parquet output " + transformName, e); + } + } + + @ProcessElement + public void process(@Element HopRow row, OutputReceiver<GenericRecord> output) { + readCounter.inc(); + try { + GenericRecord record = new GenericData.Record(schema); + Object[] values = row.getRow(); + for (int i = 0; i < schema.getFields().size(); i++) { + Schema.Field field = schema.getFields().get(i); + record.put(field.name(), value(values, indexes[i], field.schema())); + } + output.output(record); + outputCounter.inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException( + "Unable to convert a row for Parquet output " + transformName, e); + } + } + + private Object value(Object[] values, int index, Schema fieldSchema) throws Exception { + if (rowMeta.isNull(values, index)) { + return null; + } + Schema type = fieldSchema; + if (type.getType() == Schema.Type.UNION) { + for (Schema candidate : type.getTypes()) { + if (candidate.getType() != Schema.Type.NULL) { + type = candidate; + break; + } + } + } + return switch (type.getType()) { + case LONG -> longValue(values, index); + case DOUBLE -> rowMeta.getNumber(values, index); + case BOOLEAN -> rowMeta.getBoolean(values, index); + case STRING -> rowMeta.getString(values, index); + case BYTES -> rowMeta.getBinary(values, index); Review Comment: **[bug]** A BigNumber with a length is declared `decimal(length, precision)` on BYTES by `ParquetOutput.avroType`, but this fills it with `rowMeta.getBinary(values, index)`. For a BigNumber that is not the unscaled two's-complement value a decimal needs, so the column holds the wrong bytes. The local writer uses `setScale(precision).unscaledValue().toByteArray()`. **Suggestion:** for a `decimal` logical type write the unscaled bytes of the scaled `BigDecimal` (as a `ByteBuffer`), and keep `getBinary` for `TYPE_BINARY` only. ########## plugins/engines/beam/src/test/java/org/apache/hop/beam/pipeline/handler/BeamParquetFileIOTest.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.beam.pipeline.handler; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import org.apache.avro.Schema; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PCollection; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.beam.core.coder.HopRowCoder; +import org.apache.hop.beam.pipeline.HopPipelineMetaToBeamPipelineConverter; +import org.apache.hop.beam.util.BeamConst; +import org.apache.hop.core.HopEnvironment; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.logging.LogChannel; +import org.apache.hop.core.plugins.PluginRegistry; +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.ValueMetaInteger; +import org.apache.hop.core.row.value.ValueMetaString; +import org.apache.hop.core.variables.Variables; +import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider; +import org.apache.hop.parquet.transforms.input.ParquetField; +import org.apache.hop.parquet.transforms.input.ParquetInputMeta; +import org.apache.hop.parquet.transforms.output.ParquetOutput; +import org.apache.hop.parquet.transforms.output.ParquetOutputMeta; +import org.apache.hop.parquet.transforms.output.ParquetPartitionField; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.TransformMeta; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** Issue #2357: Parquet file input and output on the Beam Direct runner. */ +class BeamParquetFileIOTest { + + private static final List<String> SEEN = Collections.synchronizedList(new ArrayList<>()); + + @TempDir Path tempDir; + + @BeforeAll + static void initHopEnvironment() throws Exception { + HopEnvironment.init(); + PluginRegistry.init(); + } + + @Test + void parquetPluginIdsAreExplicitBeamHandlers() { + assertTrue( + HopPipelineMetaToBeamPipelineConverter.EXPLICIT_HANDLER_PLUGIN_IDS.contains( + BeamConst.STRING_PARQUET_FILE_OUTPUT_PLUGIN_ID)); + assertTrue( + HopPipelineMetaToBeamPipelineConverter.EXPLICIT_HANDLER_PLUGIN_IDS.contains( + BeamConst.STRING_PARQUET_FILE_INPUT_PLUGIN_ID)); + } + + @Test + void avroSchemaIsTheHopRecord() throws Exception { + ParquetOutputMeta meta = fileMeta(tempDir.resolve("schema.parquet")); + List<org.apache.hop.parquet.transforms.output.ParquetField> fields = meta.getFields(); + List<IValueMeta> types = List.of(new ValueMetaString("name"), new ValueMetaInteger("id")); + Schema schema = ParquetOutput.avroSchema(fields, types); + assertEquals("ApacheHopParquetSchema", schema.getName()); + assertNotNull(schema.getField("name")); + assertNotNull(schema.getField("id")); + assertEquals(Schema.Type.UNION, schema.getField("id").schema().getType()); + } + + @Test + void writeThenReadKeepsTheIncomingRow() throws Exception { Review Comment: **[suggestion]** The round trip only uses String and Integer columns. Adding a Date, Timestamp, BigNumber (with a length) and Binary column here would have caught the timestamp, decimal and read-side type issues above. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/HopRowToParquetFn.java: ########## @@ -0,0 +1,126 @@ +/* + * 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.beam.core.fn; + +import java.util.Date; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.hop.beam.core.BeamHop; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.pipeline.Pipeline; + +/** Maps one Hop row onto the flat Avro record ParquetIO writes. */ +@RequiredArgsConstructor +public class HopRowToParquetFn extends DoFn<HopRow, GenericRecord> { + private final String transformName; + private final String rowMetaJson; + private final String schemaJson; + private final List<String> sourceFields; + + private transient Schema schema; + private transient IRowMeta rowMeta; + private transient int[] indexes; + private transient Counter readCounter; + private transient Counter outputCounter; + private transient Counter errorCounter; + + @Setup + public void setup() { + readCounter = Metrics.counter(Pipeline.METRIC_NAME_READ, transformName); + outputCounter = Metrics.counter(Pipeline.METRIC_NAME_OUTPUT, transformName); + errorCounter = Metrics.counter(Pipeline.METRIC_NAME_ERROR, transformName); + try { + BeamHop.init(); + schema = new Schema.Parser().parse(schemaJson); + rowMeta = JsonRowMeta.fromJson(rowMetaJson); + indexes = new int[sourceFields.size()]; + for (int i = 0; i < sourceFields.size(); i++) { + indexes[i] = rowMeta.indexOfValue(sourceFields.get(i)); + if (indexes[i] < 0) { + throw new HopRuntimeException("Field [" + sourceFields.get(i) + "] was not found"); + } + } + Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException("Unable to initialize Parquet output " + transformName, e); + } + } + + @ProcessElement + public void process(@Element HopRow row, OutputReceiver<GenericRecord> output) { + readCounter.inc(); + try { + GenericRecord record = new GenericData.Record(schema); + Object[] values = row.getRow(); + for (int i = 0; i < schema.getFields().size(); i++) { + Schema.Field field = schema.getFields().get(i); + record.put(field.name(), value(values, indexes[i], field.schema())); + } + output.output(record); + outputCounter.inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException( + "Unable to convert a row for Parquet output " + transformName, e); + } + } + + private Object value(Object[] values, int index, Schema fieldSchema) throws Exception { + if (rowMeta.isNull(values, index)) { + return null; + } + Schema type = fieldSchema; + if (type.getType() == Schema.Type.UNION) { + for (Schema candidate : type.getTypes()) { + if (candidate.getType() != Schema.Type.NULL) { + type = candidate; + break; + } + } + } + return switch (type.getType()) { + case LONG -> longValue(values, index); + case DOUBLE -> rowMeta.getNumber(values, index); + case BOOLEAN -> rowMeta.getBoolean(values, index); + case STRING -> rowMeta.getString(values, index); + case BYTES -> rowMeta.getBinary(values, index); + default -> + throw new HopRuntimeException("Parquet output cannot write Avro type " + type.getType()); + }; + } + + private Object longValue(Object[] values, int index) throws Exception { + if (rowMeta.getValueMeta(index).getType() == IValueMeta.TYPE_DATE + || rowMeta.getValueMeta(index).getType() == IValueMeta.TYPE_TIMESTAMP) { + Date date = rowMeta.getDate(values, index); + return date == null ? null : date.getTime(); Review Comment: **[bug]** Timestamps are written 1000x too small. `ParquetOutput.avroSchema` declares `TYPE_TIMESTAMP` as `timestamp-micros`, but this returns `date.getTime()` (milliseconds) for both Date and Timestamp. The local writer (`ParquetWriteSupport`) writes microseconds, including the nanoseconds: `Math.floorDiv(ts.getTime(), 1000L) * 1_000_000L + ts.getNanos() / 1_000L`. Date is `timestamp-millis`, so only the Timestamp case is wrong. **Suggestion:** convert `TYPE_TIMESTAMP` to microseconds the same way as `ParquetWriteSupport`, or call a shared helper. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/handler/BeamParquetInputTransformHandler.java: ########## @@ -0,0 +1,175 @@ +/* + * 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.beam.pipeline.handler; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.apache.avro.Schema; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.transforms.Count; +import org.apache.beam.sdk.transforms.Flatten; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollectionList; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.beam.core.coder.HopRowCoder; +import org.apache.hop.beam.core.fn.EmptyToNullRowFn; +import org.apache.hop.beam.core.fn.ParquetColumnMapping; +import org.apache.hop.beam.core.fn.ReadParquetFileFn; +import org.apache.hop.beam.engines.IBeamPipelineEngineRunConfiguration; +import org.apache.hop.beam.pipeline.IBeamPipelineTransformHandler; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.logging.ILogChannel; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.parquet.transforms.input.ParquetField; +import org.apache.hop.parquet.transforms.input.ParquetInputMeta; +import org.apache.hop.parquet.transforms.output.ParquetOutput; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.TransformMeta; + +/** + * Issue #2357: Parquet file input on Beam. + * + * <p>Each incoming row names a file. The configured columns are appended to that row. The field + * list or a metadata filename is required, because the output row layout has to exist when the + * graph is built. + */ +public class BeamParquetInputTransformHandler extends BeamBaseTransformHandler + implements IBeamPipelineTransformHandler { + + @Override + public boolean isInput() { + return false; + } + + @Override + public boolean isOutput() { + return false; + } + + @Override + public void handleTransform( + ILogChannel log, + IVariables variables, + String runConfigurationName, + IBeamPipelineEngineRunConfiguration runConfiguration, + String dataSamplersJson, + IHopMetadataProvider metadataProvider, + PipelineMeta pipelineMeta, + TransformMeta transformMeta, + Map<String, PCollection<HopRow>> transformCollectionMap, + Pipeline pipeline, + IRowMeta rowMeta, + List<TransformMeta> previousTransforms, + PCollection<HopRow> input, + String parentLogChannelId) + throws HopException { + + if (input == null || previousTransforms == null || previousTransforms.isEmpty()) { + throw new HopException( + "Parquet input '" + transformMeta.getName() + "' needs one incoming transform"); + } + ParquetInputMeta meta = new ParquetInputMeta(); + loadTransformMetadata(meta, transformMeta, metadataProvider, pipelineMeta); + String filenameField = variables.resolve(meta.getFilenameField()); + if (Utils.isEmpty(filenameField)) { + throw new HopException( + "Parquet input '" + transformMeta.getName() + "' has no filename field"); + } + if (rowMeta == null || rowMeta.indexOfValue(filenameField) < 0) { + throw new HopException( + "Parquet input '" + + transformMeta.getName() + + "' cannot find filename field '" + + filenameField + + "'"); + } + + List<org.apache.hop.parquet.transforms.output.ParquetField> schemaFields = new ArrayList<>(); + List<IValueMeta> valueMetas = new ArrayList<>(); + List<ParquetColumnMapping> columns = new ArrayList<>(); + if (meta.getFields() != null && !meta.getFields().isEmpty()) { + for (ParquetField field : meta.getFields()) { + IValueMeta valueMeta = field.createValueMeta(); + String column = + Utils.isEmpty(field.getSourceField()) ? field.getTargetField() : field.getSourceField(); + schemaFields.add(new org.apache.hop.parquet.transforms.output.ParquetField(column, column)); + valueMetas.add(valueMeta); + columns.add(new ParquetColumnMapping(column, valueMeta.getType())); + } + } else if (!Utils.isEmpty(meta.getMetadataFilename())) { + IRowMeta fileRow = + ParquetInputMeta.extractRowMeta(variables, variables.resolve(meta.getMetadataFilename())); + for (int i = 0; i < fileRow.size(); i++) { + IValueMeta valueMeta = fileRow.getValueMeta(i); + schemaFields.add( + new org.apache.hop.parquet.transforms.output.ParquetField( + valueMeta.getName(), valueMeta.getName())); + valueMetas.add(valueMeta); + columns.add(new ParquetColumnMapping(valueMeta.getName(), valueMeta.getType())); + } + } + if (columns.isEmpty()) { + throw new HopException( + "Parquet input '" + + transformMeta.getName() + + "' needs a field list or a metadata filename on Beam"); + } + + Schema schema = ParquetOutput.avroSchema(schemaFields, valueMetas); + PCollection<HopRow> read = + input + .apply( + transformMeta.getName() + " Read Parquet", + ParDo.of( + new ReadParquetFileFn( + transformMeta.getName(), + JsonRowMeta.toJson(rowMeta), + filenameField, + schema.toString(), + columns))) + .setCoder(new HopRowCoder()); + + if (meta.isSendingNullsRowWhenEmpty()) { + int width = rowMeta.size() + columns.size(); + PCollection<HopRow> whenEmpty = + input + .apply(Count.globally()) Review Comment: **[suggestion]** Two things on the empty-input branch: - `Count.globally()` and the `EmptyToNullRowFn` ParDo are applied without a name, so two Parquet inputs in one pipeline get duplicate step names. Prefix them with the transform name, as the read step does. - `Count.globally()` throws while the graph is built when the input is not in the global window (streaming). A guard or a sentence in the docs would help. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/ReadParquetFileFn.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.beam.core.fn; + +import java.util.Arrays; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.hadoop.conf.Configuration; +import org.apache.hop.beam.core.BeamHop; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.core.exception.HopRuntimeException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.pipeline.Pipeline; +import org.apache.parquet.avro.AvroParquetReader; +import org.apache.parquet.avro.AvroReadSupport; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.io.LocalInputFile; + +/** + * Reads the Parquet file named by one incoming field and appends the configured columns. + * + * <p>{@code ParquetIO.readFiles} returns records without the path, so the source row would be lost. + * This reads the file on the worker that holds the row, with the same Avro projection {@code + * ParquetIO.read(schema)} would use. + */ +@RequiredArgsConstructor +public class ReadParquetFileFn extends DoFn<HopRow, HopRow> { + private final String transformName; + private final String rowMetaJson; + private final String filenameField; + private final String schemaJson; + private final List<ParquetColumnMapping> columns; + + private transient IRowMeta rowMeta; + private transient int filenameIndex; + private transient int inputSize; + private transient Schema schema; + private transient Counter readCounter; + private transient Counter outputCounter; + private transient Counter errorCounter; + + @Setup + public void setup() { + readCounter = Metrics.counter(Pipeline.METRIC_NAME_READ, transformName); + outputCounter = Metrics.counter(Pipeline.METRIC_NAME_OUTPUT, transformName); + errorCounter = Metrics.counter(Pipeline.METRIC_NAME_ERROR, transformName); + try { + BeamHop.init(); + rowMeta = JsonRowMeta.fromJson(rowMetaJson); + filenameIndex = rowMeta.indexOfValue(filenameField); + if (filenameIndex < 0) { + throw new HopRuntimeException("Filename field '" + filenameField + "' was not found"); + } + inputSize = rowMeta.size(); + schema = new Schema.Parser().parse(schemaJson); + Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc(); + } catch (Exception e) { + errorCounter.inc(); + throw new HopRuntimeException("Unable to initialize Parquet input " + transformName, e); + } + } + + @ProcessElement + public void process(@Element HopRow row, OutputReceiver<HopRow> output) { + try { + String filename = rowMeta.getString(row.getRow(), filenameIndex); + if (filename == null || filename.isEmpty()) { + return; + } + Configuration configuration = new Configuration(false); + AvroReadSupport.setRequestedProjection(configuration, schema); + try (ParquetReader<GenericRecord> reader = + AvroParquetReader.<GenericRecord>builder( + new LocalInputFile(java.nio.file.Path.of(filename))) + .withConf(configuration) + .build()) { + GenericRecord record; + while ((record = reader.read()) != null) { + Object[] copy = Arrays.copyOf(row.getRow(), inputSize + columns.size()); + for (int i = 0; i < columns.size(); i++) { + copy[inputSize + i] = + hopValue(record.get(columns.get(i).getColumnName()), columns.get(i)); + } + output.output(new HopRow(copy)); + readCounter.inc(); + outputCounter.inc(); Review Comment: **[suggestion]** Each record is counted as `READ` and `OUTPUT`. By Hop's meaning, rows passed on to the next transform are lines *written* (the other Beam source functions increment `METRIC_NAME_WRITTEN` per row); `OUTPUT` is for rows written to an external file or system. The writer side (`READ` + `OUTPUT`) is right. ########## plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/handler/BeamParquetOutputTransformHandler.java: ########## @@ -0,0 +1,268 @@ +/* + * 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.beam.pipeline.handler; + +import java.io.File; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.Map; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.extensions.avro.coders.AvroCoder; +import org.apache.beam.sdk.io.Compression; +import org.apache.beam.sdk.io.FileIO; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.PaneInfo; +import org.apache.beam.sdk.values.PCollection; +import org.apache.commons.vfs2.FileObject; +import org.apache.hop.beam.core.HopRow; +import org.apache.hop.beam.core.fn.HopRowToParquetFn; +import org.apache.hop.beam.engines.IBeamPipelineEngineRunConfiguration; +import org.apache.hop.beam.pipeline.IBeamPipelineTransformHandler; +import org.apache.hop.core.Const; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.logging.ILogChannel; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.JsonRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.core.vfs.HopVfs; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.parquet.transforms.output.ParquetField; +import org.apache.hop.parquet.transforms.output.ParquetOutput; +import org.apache.hop.parquet.transforms.output.ParquetOutputMeta; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.TransformMeta; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; + +/** + * Issue #2357: Parquet file output on Beam. + * + * <p>One file, written with {@code FileIO.write().via(ParquetIO.sink(schema))}. Date and time are + * resolved when the pipeline is submitted. Copy number and split number are not extra Beam shards. + * Partitioned output and a field with its own Parquet type are refused. + */ +public class BeamParquetOutputTransformHandler extends BeamBaseTransformHandler + implements IBeamPipelineTransformHandler { + + @Override + public boolean isInput() { + return false; + } + + @Override + public boolean isOutput() { + return false; + } + + @Override + public void handleTransform( + ILogChannel log, + IVariables variables, + String runConfigurationName, + IBeamPipelineEngineRunConfiguration runConfiguration, + String dataSamplersJson, + IHopMetadataProvider metadataProvider, + PipelineMeta pipelineMeta, + TransformMeta transformMeta, + Map<String, PCollection<HopRow>> transformCollectionMap, + Pipeline pipeline, + IRowMeta rowMeta, + List<TransformMeta> previousTransforms, + PCollection<HopRow> input, + String parentLogChannelId) + throws HopException { + + if (input == null || previousTransforms == null || previousTransforms.isEmpty()) { + throw new HopException( + "Parquet output '" + transformMeta.getName() + "' needs one incoming transform"); + } + ParquetOutputMeta meta = new ParquetOutputMeta(); + loadTransformMetadata(meta, transformMeta, metadataProvider, pipelineMeta); + if (meta.isPartitioning()) { + throw new HopException( + "Parquet output '" + transformMeta.getName() + "' cannot partition files on Beam"); + } + if (rowMeta == null) { + throw new HopException( + "Parquet output '" + transformMeta.getName() + "' has no incoming row layout"); + } + + List<ParquetField> fields = new ArrayList<>(); + List<IValueMeta> valueMetas = new ArrayList<>(); + List<String> sourceFields = new ArrayList<>(); + if (meta.getFields() != null) { + for (ParquetField field : meta.getFields()) { + if (field == null) { + continue; + } + if (field.parquetFieldType() != null) { + throw new HopException( + "Parquet output '" + + transformMeta.getName() + + "' cannot write Parquet type '" + + field.getParquetType() + + "' on Beam"); + } + if (rowMeta.indexOfValue(field.getSourceFieldName()) < 0) { + throw new HopException( + "Field [" + field.getSourceFieldName() + "] couldn't be found in the input stream!"); + } + ParquetField named = new ParquetField(field); + if (Utils.isEmpty(named.getTargetFieldName())) { + named.setTargetFieldName(named.getSourceFieldName()); + } + fields.add(named); + valueMetas.add(rowMeta.getValueMeta(rowMeta.indexOfValue(field.getSourceFieldName()))); + sourceFields.add(field.getSourceFieldName()); + } + } + if (fields.isEmpty()) { + throw new HopException( + "Parquet output '" + transformMeta.getName() + "' has no output fields"); + } + + Schema schema = ParquetOutput.avroSchema(fields, valueMetas); Review Comment: **[suggestion]** `withParquetOnlyTypes` only runs on the local engine. Locally a JSON field becomes BINARY with the json annotation and a UUID field FIXED_LEN_BYTE_ARRAY(16) with the uuid annotation. On Beam both are written as plain strings, so the same pipeline writes a different file schema. The explicit Parquet type is refused, but these conversions are automatic. **Suggestion:** apply the same conversion on Beam, or refuse JSON and UUID fields here with a clear message, and mention it on the docs page. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
