KYLIN-3546 Add column which belongs to root fact table in join relati…

…on but lost
chenzhx · Sep 24, 2018 · 98833e0 · 98833e0
1 parent f37be33
commit 98833e0
Showing 1 changed file with 23 additions and 1 deletion.
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaInputBase.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaInputBase.java
@@ -18,8 +18,11 @@
 
 package org.apache.kylin.source.kafka;
 
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Set;
 
+import com.google.common.collect.Sets;
 import org.apache.kylin.cube.CubeSegment;
 import org.apache.kylin.cube.model.CubeDesc;
 import org.apache.kylin.engine.mr.JobBuilderSupport;
@@ -32,7 +35,10 @@
 import org.apache.kylin.metadata.model.DataModelDesc;
 import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
 import org.apache.kylin.metadata.model.ISegment;
+import org.apache.kylin.metadata.model.JoinDesc;
+import org.apache.kylin.metadata.model.JoinTableDesc;
 import org.apache.kylin.metadata.model.SegmentRange;
+import org.apache.kylin.metadata.model.TableRef;
 import org.apache.kylin.metadata.model.TblColRef;
 import org.apache.kylin.source.hive.CreateFlatHiveTableStep;
 import org.apache.kylin.source.hive.GarbageCollectionStep;
@@ -89,7 +95,23 @@ public DataModelDesc getDataModel() {
 
             @Override
             public List<TblColRef> getAllColumns() {
-                return flatDesc.getFactColumns();
+                final Set<TblColRef> factTableColumnSet = Sets.newHashSet();
+                TableRef rootFactTable = getDataModel().getRootFactTable();
+                for (TblColRef colRef : flatDesc.getAllColumns()) {
+                    if (colRef.getTableRef().equals(rootFactTable)) {
+                        factTableColumnSet.add(colRef);
+                    }
+                }
+                // Add column which belongs to root fact table in join relation but lost
+                for (JoinTableDesc joinTableDesc : getDataModel().getJoinTables()) {
+                    JoinDesc join = joinTableDesc.getJoin();
+                    for (TblColRef colRef : join.getForeignKeyColumns()) {
+                        if (colRef.getTableRef().equals(rootFactTable)) {
+                            factTableColumnSet.add(colRef);
+                        }
+                    }
+                }
+                return new LinkedList<>(factTableColumnSet);
             }
 
             @Override