local commit

commit e54108864f385c4d2a89d4f1b889082bd947904c
Author: Sivabalan Narayanan <sivabala@uber.com>
Date:   Wed Jul 22 11:17:34 2020 -0400

    Adding tests for HoodieInternalRow and HoodieInternalWriteStatus

diff --git a/hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java b/hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
index 7e54e542..87a117bb 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
+++ b/hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
@@ -64,10 +64,9 @@ public class HoodieInternalWriteStatus implements Serializable {
 
   public void markFailure(String recordKey, Throwable t) {
     if (failedRecordKeys.isEmpty() || (random.nextDouble() <= failureFraction)) {
-      // Guaranteed to have at-least one error
-      //failedRows.add(new Tuple3<>(row, row.getAs(recordKeyProp), t));
       failedRecordKeys.add(Pair.of(recordKey, t));
     }
+    totalRecords++;
   }
 
   public boolean hasErrors() {
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/model/HoodieInternalRow.java b/hudi-client/src/main/java/org/apache/hudi/client/model/HoodieInternalRow.java
index 894ba69e..ccf2b907 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/model/HoodieInternalRow.java
+++ b/hudi-client/src/main/java/org/apache/hudi/client/model/HoodieInternalRow.java
@@ -223,7 +223,7 @@ public class HoodieInternalRow extends InternalRow {
     if (ordinal < HoodieRecord.HOODIE_META_COLUMNS.size()) {
       return UTF8String.fromBytes(getHoodieColumnVal(ordinal).getBytes());
     }
-    return row.get(ordinal, dataType);
+    return row.get(ordinal - HoodieRecord.HOODIE_META_COLUMNS.size(), dataType);
   }
 
   @Override
diff --git a/hudi-client/src/test/java/org/apache/hudi/client/TestHoodieInternalWriteStatus.java b/hudi-client/src/test/java/org/apache/hudi/client/TestHoodieInternalWriteStatus.java
new file mode 100644
index 00000000..4e749414
--- /dev/null
+++ b/hudi-client/src/test/java/org/apache/hudi/client/TestHoodieInternalWriteStatus.java
@@ -0,0 +1,85 @@
+/*
+ * 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.hudi.client;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests {@link HoodieInternalWriteStatus}.
+ */
+public class TestHoodieInternalWriteStatus {
+
+  @Test
+  public void testFailureFraction() {
+    HoodieInternalWriteStatus status = new HoodieInternalWriteStatus(true, 0.1);
+    String fileId = UUID.randomUUID().toString();
+    String partitionPath = UUID.randomUUID().toString();
+    status.setFileId(fileId);
+    status.setPartitionPath(partitionPath);
+    Throwable t = new Exception("some error in writing");
+    for (int i = 0; i < 1000; i++) {
+      status.markFailure(UUID.randomUUID().toString(), t);
+    }
+    // verification
+    assertEquals(fileId, status.getFileId());
+    assertEquals(partitionPath, status.getPartitionPath());
+    assertTrue(status.getFailedRecordKeys().size() > 0);
+    assertTrue(status.getFailedRecordKeys().size() < 150); // 150 instead of 100, to prevent flaky test
+    assertTrue(status.hasErrors());
+  }
+
+  @Test
+  public void testSuccessRecordTracking() {
+    boolean[] vals = {true, false};
+    for (boolean trackSuccess : vals) {
+      HoodieInternalWriteStatus status = new HoodieInternalWriteStatus(trackSuccess, 1.0);
+      String fileId = UUID.randomUUID().toString();
+      String partitionPath = UUID.randomUUID().toString();
+      Throwable t = new Exception("some error in writing");
+      for (int i = 0; i < 1000; i++) {
+        status.markSuccess(UUID.randomUUID().toString());
+        status.markFailure(UUID.randomUUID().toString(), t);
+      }
+      // verification
+      assertEquals(fileId, status.getFileId());
+      assertEquals(partitionPath, status.getPartitionPath());
+      assertEquals(1000, status.getFailedRecordKeys().size());
+      assertTrue(status.hasErrors());
+      if (trackSuccess) {
+        assertEquals(1000, status.getSuccessRecordKeys().size());
+      } else {
+        assertTrue(status.getSuccessRecordKeys().isEmpty());
+      }
+      assertEquals(2000, status.getTotalRecords());
+    }
+  }
+
+  @Test
+  public void testGlobalError() {
+    HoodieInternalWriteStatus status = new HoodieInternalWriteStatus(true, 0.1);
+    Throwable t = new Exception("some error in writing");
+    status.setGlobalError(t);
+    assertEquals(t, status.getGlobalError());
+  }
+}
diff --git a/hudi-client/src/test/java/org/apache/hudi/client/model/TestHoodieInternalRow.java b/hudi-client/src/test/java/org/apache/hudi/client/model/TestHoodieInternalRow.java
new file mode 100644
index 00000000..23c2ac37
--- /dev/null
+++ b/hudi-client/src/test/java/org/apache/hudi/client/model/TestHoodieInternalRow.java
@@ -0,0 +1,158 @@
+/*
+ * 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.hudi.client.model;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.types.DataTypes;
+import org.junit.jupiter.api.Test;
+
+import java.util.Random;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TestHoodieInternalRow {
+
+  private final static Random RANDOM = new Random();
+  private final static int integerIndex = 0;
+  private final static int stringIndex = 1;
+  private final static int booleanIndex = 2;
+  private final static int shortIndex = 3;
+  private final static int byteIndex = 4;
+  private final static int longIndex = 5;
+  private final static int floatIndex = 6;
+  private final static int doubleIndex = 7;
+  private final static int decimalIndex = 8;
+  private final static int binaryIndex = 9;
+  private final static int structIndex = 10;
+  private final static int arrayIndex = 11;
+  private final static int mapIndex = 12;
+
+  @Test
+  public void testGet() {
+    Object[] values = getRandomValue(true);
+
+    InternalRow row = new GenericInternalRow(values);
+    HoodieInternalRow hoodieInternalRow = new HoodieInternalRow("commitTime", "commitSeqNo", "recordKey", "partitionPath", "fileName", row);
+
+    assertEquals("commitTime", hoodieInternalRow.get(0, DataTypes.StringType).toString());
+    assertEquals("commitSeqNo", hoodieInternalRow.get(1, DataTypes.StringType).toString());
+    assertEquals("recordKey", hoodieInternalRow.get(2, DataTypes.StringType).toString());
+    assertEquals("partitionPath", hoodieInternalRow.get(3, DataTypes.StringType).toString());
+    assertEquals("fileName", hoodieInternalRow.get(4, DataTypes.StringType).toString());
+    assertEquals(values[integerIndex], hoodieInternalRow.getInt(integerIndex));
+    assertEquals(values[integerIndex], hoodieInternalRow.get(5 + integerIndex, DataTypes.IntegerType));
+    assertEquals(values[booleanIndex], hoodieInternalRow.getBoolean(booleanIndex));
+    assertEquals(values[booleanIndex], hoodieInternalRow.get(5 + booleanIndex, DataTypes.BooleanType));
+    assertEquals(values[shortIndex], hoodieInternalRow.getShort(shortIndex));
+    assertEquals(values[shortIndex], hoodieInternalRow.get(5 + shortIndex, DataTypes.ShortType));
+    assertEquals(values[byteIndex], hoodieInternalRow.getByte(byteIndex));
+    assertEquals(values[byteIndex], hoodieInternalRow.get(5 + byteIndex, DataTypes.ByteType));
+    assertEquals(values[longIndex], hoodieInternalRow.getLong(longIndex));
+    assertEquals(values[longIndex], hoodieInternalRow.get(5 + longIndex, DataTypes.LongType));
+    assertEquals(values[floatIndex], hoodieInternalRow.getFloat(floatIndex));
+    assertEquals(values[floatIndex], hoodieInternalRow.get(5 + floatIndex, DataTypes.FloatType));
+    assertEquals(values[doubleIndex], hoodieInternalRow.getDouble(doubleIndex));
+    assertEquals(values[doubleIndex], hoodieInternalRow.get(5 + doubleIndex, DataTypes.DoubleType));
+    assertEquals(values[binaryIndex], hoodieInternalRow.getBinary(binaryIndex));
+    assertEquals(values[binaryIndex], hoodieInternalRow.get(5 + binaryIndex, DataTypes.BinaryType));
+
+    // assertEquals(values[decimalIndex], hoodieInternalRow.getDecimal(decimalIndex, 10, 100));
+    assertEquals(values[stringIndex].toString(), hoodieInternalRow.get(5 + stringIndex, DataTypes.StringType));
+    //assertEquals(values[stringIndex].toString(), hoodieInternalRow.getString(5 + stringIndex));
+
+    assertEquals(values[structIndex], hoodieInternalRow.getStruct(structIndex, 13));
+    //assertEquals(values[arrayIndex], hoodieInternalRow.getArray(arrayIndex));
+    //assertEquals(values[mapIndex], hoodieInternalRow.getMap(mapIndex));
+    // assertEquals(values[structIndex], hoodieInternalRow.get(structIndex, DataTypes.));
+  }
+
+  private Object[] getRandomValue(boolean withStructType) {
+    Object[] values = new Object[13];
+    values[integerIndex] = RANDOM.nextInt();
+    values[stringIndex] = UUID.randomUUID().toString();
+    values[booleanIndex] = RANDOM.nextBoolean();
+    values[shortIndex] = (short) RANDOM.nextInt(2);
+    byte[] bytes = new byte[1];
+    RANDOM.nextBytes(bytes);
+    values[byteIndex] = bytes[0];
+    values[longIndex] = RANDOM.nextLong();
+    values[floatIndex] = RANDOM.nextFloat();
+    values[doubleIndex] = RANDOM.nextDouble();
+    // TODO fix decimal type.
+    values[decimalIndex] = RANDOM.nextFloat();
+    bytes = new byte[20];
+    RANDOM.nextBytes(bytes);
+    values[binaryIndex] = bytes;
+    if (withStructType) {
+      Object[] structField = getRandomValue(false);
+      values[structIndex] = new GenericInternalRow(structField);
+    }
+    /*Integer[] arrayField = new Integer[3];
+
+    Array<Integer> arrayF = new Array<Integer>(3);
+    arrayF.update(0, RANDOM.nextInt());
+    arrayF.update(1, RANDOM.nextInt());
+    arrayF.update(2, RANDOM.nextInt());
+
+    //ArrayData arrayData = ArrayData.toArrayData(arrayF);
+    arrayField[0] = RANDOM.nextInt();
+    arrayField[1] = RANDOM.nextInt();
+    arrayField[2] = RANDOM.nextInt();
+
+    ArrayType arrayType = new ArrayType(DataTypes.StringType, false);
+    //arrayType.
+
+
+    //values[arrayIndex] = ArrayData.toArrayData(arrayF);
+
+    values[arrayIndex] = arrayField;
+
+    Array<String> arrayKeys = new Array(1);
+    arrayKeys.update(0, "abc");
+
+    Array<String> arrayValues = new Array(1);
+    arrayValues.update(0, "def");
+
+    MapData mapData = new ArrayBasedMapData(ArrayData.toArrayData(arrayKeys), ArrayData.toArrayData(arrayValues));
+
+    values[mapIndex] = mapData;
+    */
+    return values;
+  }
+
+  @Test
+  public void testUpdate() {
+    Object[] values = new Object[2];
+    values[0] = 1;
+    values[1] = "abc";
+
+    InternalRow row = new GenericInternalRow(values);
+    HoodieInternalRow hoodieInternalRow = new HoodieInternalRow("commitTime", "commitSeqNo", "recordKey", "partitionPath", "fileName", row);
+
+    assertEquals("commitTime", hoodieInternalRow.get(0, DataTypes.StringType).toString());
+    assertEquals("commitSeqNo", hoodieInternalRow.get(1, DataTypes.StringType).toString());
+    assertEquals("recordKey", hoodieInternalRow.get(2, DataTypes.StringType).toString());
+    assertEquals("partitionPath", hoodieInternalRow.get(3, DataTypes.StringType).toString());
+    assertEquals("fileName", hoodieInternalRow.get(4, DataTypes.StringType).toString());
+    assertEquals(1, hoodieInternalRow.getInt(0));
+    assertEquals("abc", hoodieInternalRow.getString(1));
+  }
+}