Fix type handling in Spark and Pig. (apache#49)

SinghAsDev · Dec 13, 2018 · 9fc8f0e · 9fc8f0e
1 parent 26a3e4a
commit 9fc8f0e
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 5 deletions.
diff --git a/pig/src/main/java/com/netflix/iceberg/pig/PigParquetReader.java b/pig/src/main/java/com/netflix/iceberg/pig/PigParquetReader.java
@@ -38,6 +38,7 @@
 import com.netflix.iceberg.parquet.ParquetValueReaders.StructReader;
 import com.netflix.iceberg.parquet.ParquetValueReaders.UnboxedReader;
 import com.netflix.iceberg.parquet.TypeWithSchemaVisitor;
+import com.netflix.iceberg.types.Type.TypeID;
 import com.netflix.iceberg.types.Types;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.schema.DecimalMetadata;
@@ -208,8 +209,10 @@ public ParquetValueReader<?> primitive(com.netflix.iceberg.types.Type.PrimitiveT
           case INT_8:
           case INT_16:
           case INT_32:
-            if(expected.typeId() == Types.LongType.get().typeId()) {
+            if (expected != null && expected.typeId() == Types.LongType.get().typeId()) {
               return new IntAsLongReader(desc);
+            } else {
+              return new UnboxedReader(desc);
             }
           case INT_64: return new UnboxedReader<>(desc);
           case TIMESTAMP_MILLIS: return new TimestampMillisReader(desc);
@@ -234,13 +237,20 @@ public ParquetValueReader<?> primitive(com.netflix.iceberg.types.Type.PrimitiveT
         case FIXED_LEN_BYTE_ARRAY:
         case BINARY:
           return new BytesReader(desc);
-        case BOOLEAN:
         case INT32:
-        case INT64:
+          if (expected != null && expected.typeId() == TypeID.LONG) {
+            return new IntAsLongReader(desc);
+          } else {
+            return new UnboxedReader<>(desc);
+          }
         case FLOAT:
-          if(expected.typeId() == Types.DoubleType.get().typeId()) {
+          if (expected != null && expected.typeId() == TypeID.DOUBLE) {
             return new FloatAsDoubleReader(desc);
+          } else {
+            return new UnboxedReader<>(desc);
           }
+        case BOOLEAN:
+        case INT64:
         case DOUBLE:
           return new UnboxedReader<>(desc);
         default:

diff --git a/spark/src/main/java/com/netflix/iceberg/spark/data/SparkParquetReaders.java b/spark/src/main/java/com/netflix/iceberg/spark/data/SparkParquetReaders.java
@@ -204,10 +204,15 @@ public ParquetValueReader<?> primitive(com.netflix.iceberg.types.Type.PrimitiveT
           case JSON:
           case UTF8:
             return new StringReader(desc);
-          case DATE:
           case INT_8:
           case INT_16:
           case INT_32:
+            if (expected != null && expected.typeId() == Types.LongType.get().typeId()) {
+              return new IntAsLongReader(desc);
+            } else {
+              return new UnboxedReader(desc);
+            }
+          case DATE:
           case INT_64:
           case TIMESTAMP_MICROS:
             return new UnboxedReader<>(desc);