update

Author: ArnavBalyan

parquet-column/src/main/java/org/apache/parquet/io/ValidatingRecordConsumer.java

@@ -30,8 +30,10 @@
 import java.util.ArrayDeque;
 import java.util.Arrays;
 import java.util.Deque;
+import java.util.Optional;
 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.MessageType;
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import org.apache.parquet.schema.Type;
@@ -202,6 +204,7 @@ private void validate(PrimitiveTypeName... ptypes) {
   @Override
   public void addInteger(int value) {
     validate(INT32);
+    validateUnsignedInteger(value);
     delegate.addInteger(value);
   }
 
@@ -211,6 +214,7 @@ public void addInteger(int value) {
   @Override
   public void addLong(long value) {
     validate(INT64);
+    validateUnsignedLong(value);
     delegate.addLong(value);
   }
 
@@ -249,4 +253,66 @@ public void addDouble(double value) {
     validate(DOUBLE);
     delegate.addDouble(value);
   }
+
+  private void validateUnsignedInteger(int value) {
+    Type currentType = types.peek().asGroupType().getType(fields.peek());
+    if (currentType != null && currentType.isPrimitive()) {
+      LogicalTypeAnnotation logicalType = currentType.asPrimitiveType().getLogicalTypeAnnotation();
+      if (logicalType != null) {
+        logicalType.accept(new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<Void>() {
+          @Override
+          public Optional<Void> visit(LogicalTypeAnnotation.IntLogicalTypeAnnotation intType) {
+            if (!intType.isSigned()) {
+              switch (intType.getBitWidth()) {
+                case 8:
+                  if (value < 0 || value > 255) {
+                    throw new InvalidRecordException("Value " + value
+                        + " is out of range for UINT_8 (0-255) in field "
+                        + currentType.getName());
+                  }
+                  break;
+                case 16:
+                  if (value < 0 || value > 65535) {
+                    throw new InvalidRecordException("Value " + value
+                        + " is out of range for UINT_16 (0-65535) in field "
+                        + currentType.getName());
+                  }
+                  break;
+                case 32:
+                case 64:
+                  if (value < 0) {
+                    throw new InvalidRecordException("Negative value " + value
+                        + " is not allowed for unsigned integer type "
+                        + currentType.getName());
+                  }
+                  break;
+              }
+            }
+            return Optional.empty();
+          }
+        });
+      }
+    }
+  }
+
+  private void validateUnsignedLong(long value) {
+    Type currentType = types.peek().asGroupType().getType(fields.peek());
+    if (currentType != null && currentType.isPrimitive()) {
+      LogicalTypeAnnotation logicalType = currentType.asPrimitiveType().getLogicalTypeAnnotation();
+      if (logicalType != null) {
+        logicalType.accept(new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<Void>() {
+          @Override
+          public Optional<Void> visit(LogicalTypeAnnotation.IntLogicalTypeAnnotation intType) {
+            if (!intType.isSigned()) {
+              if (value < 0) {
+                throw new InvalidRecordException("Negative value " + value
+                    + " is not allowed for unsigned integer type " + currentType.getName());
+              }
+            }
+            return Optional.empty();
+          }
+        });
+      }
+    }
+  }
 }