Updated output formats to output BigDecimal fields as numeric values …

…instead of a byte representation.

format-common/src/main/java/io/cdap/cdap/format/io/BigDecimalAwareJsonStructuredRecordDatumWriter.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.format.io;
+
+import com.google.gson.stream.JsonWriter;
+import io.cdap.cdap.api.common.Bytes;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.common.io.Encoder;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+
+/**
+ * Custom Datum Writer for Structured Records, this class writes BigDecimal values as numbers instead of byte arrays in
+ * the JSON output.
+ */
+public class BigDecimalAwareJsonStructuredRecordDatumWriter extends JsonStructuredRecordDatumWriter {
+  @Override
+  protected void encode(Encoder encoder, Schema schema, Object value) throws IOException {
+    Schema nonNullableSchema = schema.isNullable() ? schema.getNonNullable() : schema;
+    Schema.LogicalType logicalType = nonNullableSchema.getLogicalType();
+
+    if (value != null && logicalType == Schema.LogicalType.DECIMAL) {
+      BigDecimal bdValue = fromObject(value, nonNullableSchema);
+      getJsonWriter(encoder).value(bdValue);
+      return;
+    }
+
+    super.encode(encoder, schema, value);
+  }
+
+  /**
+   * Extract a BigDecimal value from a supplied object
+   * @param value object to convert into BigDecimal
+   * @param logicalTypeSchema logical type schema for this field
+   * @return value converted ingo a BigDecimal.
+   */
+  protected BigDecimal fromObject(Object value, Schema logicalTypeSchema) {
+    // Return BigDecimal as is
+    if (value instanceof BigDecimal) {
+      return (BigDecimal) value;
+    }
+
+    // Handle the value as a byte buffer
+    int scale = logicalTypeSchema.getScale();
+    if (value instanceof ByteBuffer) {
+      return new BigDecimal(new BigInteger(Bytes.toBytes((ByteBuffer) value)), scale);
+    }
+
+    // Handle the BigDecimal value
+    try {
+      return new BigDecimal(new BigInteger((byte[]) value), scale);
+    } catch (ClassCastException e) {
+      throw new ClassCastException(String.format("Field '%s' is expected to be a decimal, but is a %s.",
+                                                 logicalTypeSchema.getDisplayName(),
+                                                 value.getClass().getSimpleName()));
+    }
+  }
+
+  private JsonWriter getJsonWriter(Encoder encoder) {
+    // Type already checked in the encode method, hence assuming the casting is fine.
+    return ((JsonEncoder) encoder).getJsonWriter();
+  }
+}

format-common/src/main/java/io/cdap/cdap/format/io/BigDecimalAwareJsonWriter.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.format.io;
+
+import com.google.gson.stream.JsonWriter;
+
+import java.io.IOException;
+import java.io.Writer;
+import java.math.BigDecimal;
+
+/**
+ * JsonWriter instance which handles writing BigDecimal fields.
+ */
+public class BigDecimalAwareJsonWriter extends JsonWriter {
+  public BigDecimalAwareJsonWriter(Writer out) {
+    super(out);
+  }
+
+  @Override
+  public JsonWriter value(Number value) throws IOException {
+    if (value == null) {
+      return this.nullValue();
+    }
+
+    // Wrap BigDecimal fields in a wrapper which handles the conversion to String.
+    if (value instanceof BigDecimal) {
+      value = new BigDecimalWrapper((BigDecimal) value);
+    }
+
+    super.value(value);
+    return this;
+  }
+
+  /**
+   * Wrapper used to ensure that BigDecimals are generated as plain strings.
+   */
+  private static class BigDecimalWrapper extends Number {
+    BigDecimal wrapped;
+
+    protected BigDecimalWrapper(BigDecimal wrapped) {
+      this.wrapped = wrapped;
+    }
+
+    @Override
+    public String toString() {
+      return wrapped.toPlainString();
+    }
+
+    @Override
+    public int intValue() {
+      return wrapped.intValue();
+    }
+
+    @Override
+    public long longValue() {
+      return wrapped.longValue();
+    }
+
+    @Override
+    public float floatValue() {
+      return wrapped.floatValue();
+    }
+
+    @Override
+    public double doubleValue() {
+      return wrapped.doubleValue();
+    }
+  }
+}

format-common/src/main/java/io/cdap/plugin/format/BigDecimalAwareStructuredRecordStringConverter.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.format;
+
+import com.google.gson.stream.JsonWriter;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.format.io.BigDecimalAwareJsonStructuredRecordDatumWriter;
+import io.cdap.cdap.format.io.BigDecimalAwareJsonWriter;
+import io.cdap.cdap.format.io.JsonEncoder;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.math.BigDecimal;
+import java.util.stream.Collectors;
+
+/**
+ * Structured record converted that handles writing decimal files as numbers in the output (instead of int arrays).
+ */
+public class BigDecimalAwareStructuredRecordStringConverter {
+  private static final BigDecimalAwareJsonStructuredRecordDatumWriter JSON_DATUM_WRITER =
+    new BigDecimalAwareJsonStructuredRecordDatumWriter();
+
+  /**
+   * Converts a {@link StructuredRecord} to a json string.
+   */
+  public static String toJsonString(StructuredRecord record) throws IOException {
+    StringWriter strWriter = new StringWriter();
+    try (JsonWriter writer = new BigDecimalAwareJsonWriter(strWriter)) {
+      JSON_DATUM_WRITER.encode(record, new JsonEncoder(writer));
+      return strWriter.toString();
+    }
+  }
+
+  /**
+   * Converts a {@link StructuredRecord} to a delimited string.
+   */
+  public static String toDelimitedString(final StructuredRecord record, String delimiter) {
+    return record.getSchema().getFields().stream()
+      .map(f -> mapField(record, f))
+      .collect(Collectors.joining(delimiter));
+  }
+
+  /**
+   * Get the string representation for a given record field. BigDecimals are printed as plain strings.
+   * @param record record to process
+   * @param field field to extract
+   * @return String representing the value for this field.
+   */
+  private static String mapField(StructuredRecord record, Schema.Field field) {
+    String fieldName = field.getName();
+    Object value = record.get(fieldName);
+
+    // Return null value as empty string.
+    if (value == null) {
+      return "";
+    }
+
+    // Get the field schema.
+    Schema fieldSchema = field.getSchema();
+    if (fieldSchema.isNullable()) {
+      fieldSchema = fieldSchema.getNonNullable();
+    }
+
+    // Write decimal values as decimal strings.
+    if (fieldSchema.getLogicalType() != null && fieldSchema.getLogicalType() == Schema.LogicalType.DECIMAL) {
+      BigDecimal decimalValue = record.getDecimal(fieldName);
+
+      // Throw exception if the field is expected tu be decimal, but it could not be processed as such.
+      if (decimalValue == null) {
+        throw new IllegalArgumentException("Invalid schema for field " + fieldName + ". Decimal was expected.");
+      }
+      return decimalValue.toPlainString();
+    }
+
+    return value.toString();
+  }
+
+  private BigDecimalAwareStructuredRecordStringConverter() {
+    //inaccessible constructor for static class
+  }
+}

format-common/src/test/java/io/cdap/plugin/format/BigDecimalAwareStructuredRecordStringConverterTest.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.format;
+
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.MathContext;
+
+public class BigDecimalAwareStructuredRecordStringConverterTest {
+  private static final Schema SCHEMA = Schema.recordOf(
+    "rec",
+    Schema.Field.of("i", Schema.of(Schema.Type.INT)),
+    Schema.Field.of("f", Schema.of(Schema.Type.FLOAT)),
+    Schema.Field.of("bd", Schema.nullableOf(Schema.decimalOf(32, 7))),
+    Schema.Field.of("d", Schema.of(Schema.Type.DOUBLE)),
+    Schema.Field.of("l", Schema.of(Schema.Type.LONG)));
+
+  private static final BigDecimal BD1 =
+    new BigDecimal("12398321.8127312", MathContext.DECIMAL128).setScale(7);
+
+  private static final BigDecimal BD2 =
+    new BigDecimal("1000000000000000.0", MathContext.DECIMAL128).setScale(7);
+
+  private static final StructuredRecord RECORD1 = StructuredRecord.builder(SCHEMA)
+    .set("i", 1)
+    .set("f", 256.1f)
+    .setDecimal("bd", BD1)
+    .set("d", 12345.45678d)
+    .set("l", 123456789L)
+    .build();
+
+  private static final StructuredRecord RECORD2 = StructuredRecord.builder(SCHEMA)
+    .set("i", 1)
+    .set("f", 256.1f)
+    .setDecimal("bd", BD2)
+    .set("d", 12345.45678d)
+    .set("l", 123456789L)
+    .build();
+
+  private static final StructuredRecord RECORD3 = StructuredRecord.builder(SCHEMA)
+    .set("i", 1)
+    .set("f", 256.1f)
+    .setDecimal("bd", null)
+    .set("d", 12345.45678d)
+    .set("l", 123456789L)
+    .build();
+
+  @Test
+  public void testToJsonString1() throws IOException {
+    String output = BigDecimalAwareStructuredRecordStringConverter.toJsonString(RECORD1);
+    Assert.assertTrue(output.startsWith("{\"i\":1,"));
+    Assert.assertTrue(output.contains(",\"f\":256.1"));
+    Assert.assertTrue(output.contains(",\"bd\":12398321.8127312,"));
+    Assert.assertTrue(output.contains(",\"d\":12345.45678,"));
+    Assert.assertTrue(output.endsWith(",\"l\":123456789}"));
+  }
+
+  @Test
+  public void testToJsonString2() throws IOException {
+    String output = BigDecimalAwareStructuredRecordStringConverter.toJsonString(RECORD2);
+    Assert.assertTrue(output.startsWith("{\"i\":1,"));
+    Assert.assertTrue(output.contains(",\"f\":256.1"));
+    Assert.assertTrue(output.contains(",\"bd\":1000000000000000.0000000,"));
+    Assert.assertTrue(output.contains(",\"d\":12345.45678,"));
+    Assert.assertTrue(output.endsWith(",\"l\":123456789}"));
+  }
+
+  @Test
+  public void testToJsonString3() throws IOException {
+    String output = BigDecimalAwareStructuredRecordStringConverter.toJsonString(RECORD3);
+    Assert.assertTrue(output.startsWith("{\"i\":1,"));
+    Assert.assertTrue(output.contains(",\"f\":256.1"));
+    Assert.assertTrue(output.contains(",\"bd\":null,"));
+    Assert.assertTrue(output.contains(",\"d\":12345.45678,"));
+    Assert.assertTrue(output.endsWith(",\"l\":123456789}"));
+  }
+
+  @Test
+  public void testToDelimitedStringUsingComma() {
+    String output = BigDecimalAwareStructuredRecordStringConverter.toDelimitedString(RECORD1, ",");
+    Assert.assertTrue(output.startsWith("1,"));
+    Assert.assertTrue(output.contains(",256.1,"));
+    Assert.assertTrue(output.contains(",12398321.8127312,"));
+    Assert.assertTrue(output.contains(",12345.45678,"));
+    Assert.assertTrue(output.endsWith(",123456789"));
+  }
+
+  @Test
+  public void testToDelimitedStringUsingPipe() {
+    String output = BigDecimalAwareStructuredRecordStringConverter.toDelimitedString(RECORD1, "|");
+    Assert.assertTrue(output.startsWith("1|"));
+    Assert.assertTrue(output.contains("|256.1|"));
+    Assert.assertTrue(output.contains("|12398321.8127312|"));
+    Assert.assertTrue(output.contains("|12345.45678|"));
+    Assert.assertTrue(output.endsWith("|123456789"));
+  }
+}

format-delimited/src/main/java/io/cdap/plugin/format/delimited/output/StructuredDelimitedOutputFormat.java

@@ -19,7 +19,7 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.dataset.lib.KeyValue;
-import io.cdap.cdap.format.StructuredRecordStringConverter;
+import io.cdap.plugin.format.BigDecimalAwareStructuredRecordStringConverter;
 import io.cdap.plugin.format.output.DelegatingOutputFormat;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.NullWritable;
@@ -63,8 +63,9 @@ protected OutputFormat<NullWritable, Text> createDelegate() {
   @Override
   protected Function<StructuredRecord, KeyValue<NullWritable, Text>> getConversion(TaskAttemptContext context) {
     String delimiter = getDelimiter(context.getConfiguration());
-    return record -> new KeyValue<>(NullWritable.get(),
-                                    new Text(StructuredRecordStringConverter.toDelimitedString(record, delimiter)));
+    return record -> new KeyValue<>(
+      NullWritable.get(), new Text(BigDecimalAwareStructuredRecordStringConverter.toDelimitedString(record, delimiter))
+    );
   }
 
   @Nullable