From fb7ecfa38fe91266203a875d8a3522061ef07d18 Mon Sep 17 00:00:00 2001
From: Daniel Weeks <dweeks@apache.org>
Date: Mon, 29 Jul 2019 09:32:46 -0700
Subject: [PATCH] Fix pig partition/vavlue mapping

---
 .../org/apache/iceberg/pig/IcebergPigInputFormat.java     | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/pig/src/main/java/org/apache/iceberg/pig/IcebergPigInputFormat.java b/pig/src/main/java/org/apache/iceberg/pig/IcebergPigInputFormat.java
index ca2529b03ccb..e9b41ec8ef66 100644
--- a/pig/src/main/java/org/apache/iceberg/pig/IcebergPigInputFormat.java
+++ b/pig/src/main/java/org/apache/iceberg/pig/IcebergPigInputFormat.java
@@ -197,12 +197,16 @@ private boolean advance() throws IOException {
           if (hasJoinedPartitionColumns) {
 
             Schema readSchema = TypeUtil.selectNot(projectedSchema, idColumns);
-            Schema partitionSchema = TypeUtil.select(tableSchema, idColumns);
             Schema projectedPartitionSchema = TypeUtil.select(projectedSchema, idColumns);
 
+            Map<String, Integer> partitionSpecFieldIndexMap = Maps.newHashMap();
+            for(int i=0; i<spec.fields().size(); i++) {
+              partitionSpecFieldIndexMap.put(spec.fields().get(i).name(), i);
+            }
+
             for (Types.NestedField field : projectedPartitionSchema.columns()) {
               int tupleIndex = projectedSchema.columns().indexOf(field);
-              int partitionIndex = partitionSchema.columns().indexOf(field);
+              int partitionIndex = partitionSpecFieldIndexMap.get(field.name());
 
               Object partitionValue = file.partition().get(partitionIndex, Object.class);
               partitionValueMap.put(tupleIndex, convertPartitionValue(field.type(), partitionValue));