Fix outputoffset when build dag req (#615)

pingcap · Mar 28, 2019 · 2e9b181 · 2e9b181
1 parent ce6e612
commit 2e9b181
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 12 deletions.
diff --git a/core/src/test/scala/org/apache/spark/sql/expression/LogicalAndOr0Suite.scala b/core/src/test/scala/org/apache/spark/sql/expression/LogicalAndOr0Suite.scala
@@ -21,6 +21,7 @@ import org.apache.spark.sql.BaseInitialOnceSuite
 
 class LogicalAndOr0Suite extends BaseInitialOnceSuite {
   private val allCases = Seq[String](
+    "select tp_char, tp_smallint, tp_char, tp_smallint, tp_char from full_data_type_table  where tp_char = tp_smallint and tp_char > 0",
     "select tp_char,tp_smallint from full_data_type_table  where tp_char = tp_smallint and tp_char > 0",
     "select tp_char,id_dt from full_data_type_table  where tp_char = id_dt and tp_char > 0",
     "select tp_nvarchar,tp_tinyint from full_data_type_table  where tp_nvarchar = tp_tinyint and tp_nvarchar > 0",

diff --git a/tikv-client/src/main/java/com/pingcap/tikv/meta/TiDAGRequest.java b/tikv-client/src/main/java/com/pingcap/tikv/meta/TiDAGRequest.java
@@ -356,28 +356,32 @@ else if (col.getColumnInfo().isPrimaryKey() && tableInfo.isPkHandle()) {
       executorBuilder.setTp(ExecType.TypeTableScan);
       tblScanBuilder.setTableId(id);
       // Step1. Add columns to first executor
+      int realOffset = 0;
       for (int i = 0; i < getFields().size(); i++) {
         ColumnRef col = getFields().get(i);
-        tblScanBuilder.addColumns(col.getColumnInfo().toProto(tableInfo));
-        colOffsetInFieldMap.put(col, i);
+        // can't allow duplicated col added into executor.
+        if (!colOffsetInFieldMap.containsKey(col)) {
+          tblScanBuilder.addColumns(col.getColumnInfo().toProto(tableInfo));
+          colOffsetInFieldMap.put(col, realOffset);
+          realOffset++;
+        }
+
+        // column offset should be in accordance with fields
+        dagRequestBuilder.addOutputOffsets(colOffsetInFieldMap.getOrDefault(col, realOffset));
       }
+
       // Currently, according to TiKV's implementation, if handle
       // is needed, we should add an extra column with an ID of -1
       // to the TableScan executor
       if (isHandleNeeded()) {
         tblScanBuilder.addColumns(handleColumn);
-      }
-      dagRequestBuilder.addExecutors(executorBuilder.setTblScan(tblScanBuilder));
-
-      // column offset should be in accordance with fields
-      for (int i = 0; i < getFields().size(); i++) {
-        dagRequestBuilder.addOutputOffsets(i);
+        // if handle is needed, we should append one output offset
+        // duplicated col may exists, we need append the size of current
+        // output offset's size.
+        dagRequestBuilder.addOutputOffsets(dagRequestBuilder.getOutputOffsetsCount());
       }
 
-      // if handle is needed, we should append one output offset
-      if (isHandleNeeded()) {
-        dagRequestBuilder.addOutputOffsets(tableInfo.getColumns().size());
-      }
+      dagRequestBuilder.addExecutors(executorBuilder.setTblScan(tblScanBuilder));
     }
 
     if (!isIndexScan || (isIndexScan() && !isDoubleRead())) {

diff --git a/tikv-client/src/main/java/com/pingcap/tikv/operation/iterator/CoprocessIterator.java b/tikv-client/src/main/java/com/pingcap/tikv/operation/iterator/CoprocessIterator.java
@@ -60,6 +60,8 @@ public abstract class CoprocessIterator<T> implements Iterator<T> {
   public static CoprocessIterator<Row> getRowIterator(
       TiDAGRequest req, List<RegionTask> regionTasks, TiSession session) {
     return new DAGIterator<Row>(
+        // If index scan is a covering index, the logic is table scan
+        // so we need set isIndexScan to false.
         req.buildScan(req.isIndexScan() && !req.isDoubleRead()),
         regionTasks,
         session,