From 0526cecefc60523f7bf3858b60d22770ffe777b8 Mon Sep 17 00:00:00 2001
From: blrunner <jhjung@gruter.com>
Date: Tue, 22 Jul 2014 23:08:39 +0900
Subject: [PATCH 1/2] TAJO-968: Self-Join query (including partitioned table)
 doesn't run unexpectedly using auto broad cast join. (jaehwa)

---
 .../tajo/worker/TaskAttemptContext.java       | 26 +++++++++++-
 .../tajo/engine/query/TestJoinBroadcast.java  | 40 ++++++++++++++++++-
 2 files changed, 62 insertions(+), 4 deletions(-)

diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
index 1f0c410eb0..db4af458ea 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskAttemptContext.java
@@ -31,8 +31,10 @@
 import org.apache.tajo.engine.planner.enforce.Enforcer;
 import org.apache.tajo.engine.planner.global.DataChannel;
 import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.storage.fragment.FileFragment;
 import org.apache.tajo.storage.fragment.Fragment;
 import org.apache.tajo.storage.fragment.FragmentConvertor;
+import org.apache.tajo.util.TUtil;
 
 import java.io.File;
 import java.util.*;
@@ -234,10 +236,30 @@ public void addFragments(String tableId, FragmentProto[] fragments) {
       tableFragments = new ArrayList<FragmentProto>();
     }
 
+    List<Path> paths = fragmentToPath(tableFragments);
+
     for (FragmentProto eachFragment: fragments) {
-      tableFragments.add(eachFragment);
+      FileFragment fileFragment = FragmentConvertor.convert(FileFragment.class, eachFragment);
+      // If current attempt already has same path, we don't need to add it to fragments.
+      if (!paths.contains(fileFragment.getPath())) {
+        tableFragments.add(eachFragment);
+      }
     }
-    fragmentMap.put(tableId, tableFragments);
+
+    if (tableFragments.size() > 0) {
+      fragmentMap.put(tableId, tableFragments);
+    }
+  }
+
+  private List<Path> fragmentToPath(List<FragmentProto> tableFragments) {
+    List<Path> list = TUtil.newList();
+
+    for (FragmentProto proto : tableFragments) {
+      FileFragment fragment = FragmentConvertor.convert(FileFragment.class, proto);
+      list.add(fragment.getPath());
+    }
+
+    return list;
   }
 
   public Path getWorkDir() {
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
index e01b3c5442..a951f4bae3 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
@@ -44,9 +44,9 @@
 import java.io.File;
 import java.sql.ResultSet;
 
-import static junit.framework.TestCase.*;
+import static junit.framework.TestCase.fail;
 import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
-import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.*;
 
 @Category(IntegrationTest.class)
 public class TestJoinBroadcast extends QueryTestCaseBase {
@@ -576,4 +576,40 @@ private void createMultiFile(String tableName, int numRowsEachFile, TupleCreator
     appender.flush();
     appender.close();
   }
+
+
+  @Test
+  public final void testSelfJoin() throws Exception {
+    String tableName = CatalogUtil.normalizeIdentifier("paritioned_nation");
+    ResultSet res = executeString(
+        "create table " + tableName + " (n_name text,"
+            + "  n_comment text, n_regionkey int8) USING csv "
+            + "WITH ('csvfile.delimiter'='|')"
+            + "PARTITION BY column(n_nationkey int8)");
+    res.close();
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    res = executeString(
+        "insert overwrite into " + tableName
+            + " select n_name, n_comment, n_regionkey, n_nationkey from nation");
+    res.close();
+
+    res = executeString(
+      "select a.n_nationkey, a.n_name from nation a join nation b on a.n_nationkey = b.n_nationkey"
+      + " where a.n_nationkey in (1)");
+    String expected = resultSetToString(res);
+    res.close();
+
+    res = executeString(
+      "select a.n_nationkey, a.n_name from " + tableName + " a join "+tableName +
+      " b on a.n_nationkey = b.n_nationkey "
+      + " where a.n_nationkey in (1)");
+    String resultSetData = resultSetToString(res);
+    res.close();
+
+    assertEquals(expected, resultSetData);
+
+  }
+
+
 }

From 2b1b5fbbb651ad87bad8d7b524303cc772283a1d Mon Sep 17 00:00:00 2001
From: blrunner <blrunner@apache.org>
Date: Thu, 24 Jul 2014 11:16:22 +0900
Subject: [PATCH 2/2] Updated Change Note

---
 CHANGES | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/CHANGES b/CHANGES
index 8a1aae651b..e893944aac 100644
--- a/CHANGES
+++ b/CHANGES
@@ -97,6 +97,9 @@ Release 0.9.0 - unreleased
 
   BUG FIXES
 
+    TAJO-968: Self-Join query (including partitioned table) doesn't run unexpectedly 
+    using auto broadcast join. (jaewha)
+
     TAJO-666: java.nio.BufferOverflowException occurs when the query includes an order by 
     clause on a TEXT column. (Mai Hai Thanh via jihoon)