Skip dropped columns

src/backend/distributed/planner/local_distributed_join_planner.c

@@ -132,6 +132,11 @@
  */
 int LocalTableJoinPolicy = LOCAL_JOIN_POLICY_AUTO;
 
+/*
+ * RangeTableEntryDetails contains some information about
+ * a range table entry so that we don't need to calculate
+ * them over and over.
+ */
 typedef struct RangeTableEntryDetails
 {
 	RangeTblEntry *rangeTableEntry;
@@ -139,17 +144,33 @@ typedef struct RangeTableEntryDetails
 	bool hasConstantFilterOnUniqueColumn;
 } RangeTableEntryDetails;
 
+/*
+ * ConversionCandidates contains candidates that could
+ * be converted to a subquery. This is used as a convenience to
+ * first generate all the candidates and then choose which ones to convert.
+ */
 typedef struct ConversionCandidates
 {
 	List *distributedTableList; /* reference or distributed table */
 	List *localTableList; /* local or citus local table */
 }ConversionCandidates;
 
+
+/*
+ * IndexColumns contains the column numbers for an index.
+ * For example if there is an index on (a, b) then it will contain
+ * their column numbers (1,2).
+ */
 typedef struct IndexColumns
 {
 	List *indexColumnNos;
 }IndexColumns;
 
+/*
+ * ConversionChoice represents which conversion group
+ * to convert to a subquery. Currently we either convert all
+ * local tables, or distributed tables.
+ */
 typedef enum ConversionChoice
 {
 	CONVERT_LOCAL_TABLES = 1,
@@ -458,6 +479,13 @@ RequiredAttrNumbersForRelation(RangeTblEntry *rangeTableEntry,
 	}
 
 	PlannerInfo *plannerInfo = relationRestriction->plannerInfo;
+
+	/*
+	 * Here we used the query from plannerInfo because it has the optimizations
+	 * so that it doesn't have unnecessary columns. The original query doesn't have
+	 * some of these optimizations hence if we use it here, we don't get the
+	 * 'required' attributes.
+	 */
 	Query *queryToProcess = plannerInfo->parse;
 	int rteIndex = relationRestriction->index;
 

src/backend/distributed/planner/query_colocation_checker.c

@@ -321,6 +321,11 @@ CreateAllTargetListForRelation(Oid relationId, List *requiredAttributes)
 		Form_pg_attribute attributeTuple =
 			TupleDescAttr(relation->rd_att, attrNum - 1);
 
+		if (attributeTuple->attisdropped)
+		{
+			continue;
+		}
+
 		int resNo = attrNum;
 		if (!list_member_int(requiredAttributes, attrNum))
 		{

src/test/regress/expected/local_table_join.out

@@ -1256,7 +1256,30 @@ DEBUG:  Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS c
      2
 (1 row)
 
+CREATE TABLE local (key1 int, key2 int, key3 int);
+INSERT INTO local VALUES (1,2,3);
+ALTER TABLE local DROP column key2;
+-- verify we ignore dropped columns
+SELECT COUNT(*) FROM local JOIN distributed_table ON(key1 = key);
+DEBUG:  Wrapping relation "local" to a subquery
+DEBUG:  generating subplan XXX_1 for subquery SELECT key1 FROM local_table_join.local WHERE true
+DEBUG:  Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT local_1.key1, NULL::integer AS key3 FROM (SELECT intermediate_result.key1 FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key1 integer)) local_1) local JOIN local_table_join.distributed_table ON ((local.key1 OPERATOR(pg_catalog.=) distributed_table.key)))
+ count
+---------------------------------------------------------------------
+     1
+(1 row)
+
+SELECT * FROM local JOIN distributed_table ON(key1 = key) ORDER BY 1 LIMIT 1;
+DEBUG:  Wrapping relation "local" to a subquery
+DEBUG:  generating subplan XXX_1 for subquery SELECT key1, key3 FROM local_table_join.local WHERE true
+DEBUG:  Plan XXX query after replacing subqueries and CTEs: SELECT local.key1, local.key3, distributed_table.key, distributed_table.value, distributed_table.value_2 FROM ((SELECT local_1.key1, local_1.key3 FROM (SELECT intermediate_result.key1, intermediate_result.key3 FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key1 integer, key3 integer)) local_1) local JOIN local_table_join.distributed_table ON ((local.key1 OPERATOR(pg_catalog.=) distributed_table.key))) ORDER BY local.key1 LIMIT 1
+DEBUG:  push down of limit count: 1
+ key1 | key3 | key | value | value_2
+---------------------------------------------------------------------
+    1 |    3 |   1 | 1     |
+(1 row)
+
 RESET client_min_messages;
 \set VERBOSITY terse
 DROP SCHEMA local_table_join CASCADE;
-NOTICE:  drop cascades to 15 other objects
+NOTICE:  drop cascades to 16 other objects

src/test/regress/sql/local_table_join.sql

@@ -315,6 +315,14 @@ EXECUTE local_dist_table_join_filters(20);
 EXECUTE local_dist_table_join_filters(20);
 EXECUTE local_dist_table_join_filters(20);
 
+CREATE TABLE local (key1 int, key2 int, key3 int);
+INSERT INTO local VALUES (1,2,3);
+ALTER TABLE local DROP column key2;
+-- verify we ignore dropped columns
+SELECT COUNT(*) FROM local JOIN distributed_table ON(key1 = key);
+SELECT * FROM local JOIN distributed_table ON(key1 = key) ORDER BY 1 LIMIT 1;
+
+
 RESET client_min_messages;
 \set VERBOSITY terse
 DROP SCHEMA local_table_join CASCADE;