Fix CTE Column Pruning (#942)

* Fix nested join callback so that the bitmap is at the block level rather than using global indices.
* Save new BoundColumnRefExpressions for CTE nodes in remove_unused_columns pass so that references to them stay around while they are being used.

Author: DrTodd13

CMakeLists.txt

@@ -229,7 +229,7 @@ if (CMAKE_BUILD_TYPE STREQUAL "Debug" OR CMAKE_BUILD_TYPE STREQUAL "DebugSanitiz
   if (NOT WIN32)
     add_compile_options(
       "-g"
-      "-O1"
+      "-O0"
       "-fno-omit-frame-pointer"
     )
   else()

bodo/pandas/physical/expression.cpp

@@ -857,7 +857,8 @@ void PhysicalExpression::join_expr_batch(
     int64_t right_index_start, int64_t right_index_end) {
     for (int64_t j = right_index_start; j < right_index_end; j++) {
         for (int64_t i = left_index_start; i < left_index_end; i++) {
-            SetBitTo(match_arr, i + j,
+            SetBitTo(match_arr,
+                     (i - left_index_start) + (j - right_index_start),
                      join_expr(left_table, right_table, left_data, right_data,
                                left_null_bitmap, right_null_bitmap, i, j));
         }

bodo/pandas/vendor/duckdb/src/include/duckdb/optimizer/remove_unused_columns.hpp

@@ -84,6 +84,7 @@ class RemoveUnusedColumnsPass {
     // same CTE index by storing the addresses of the CTERef nodes
     // that we've seen for each CTE index thus far.
     unordered_map<idx_t, unordered_set<void *>> cte_ref_check;
+    vector<unique_ptr<BoundColumnRefExpression>> temp_column_refs;
 };
 // Bodo Change End
 

bodo/pandas/vendor/duckdb/src/optimizer/remove_unused_columns.cpp

@@ -548,8 +548,8 @@ void RemoveUnusedColumns::VisitOperator(LogicalOperator &op) {
                 ColumnBinding cte_binding(cte_id, i);
                 if (column_references.find(cte_binding) != column_references.end()) {
                     ColumnBinding child_binding(use_cte_root_table_index, i);
-                    BoundColumnRefExpression tempcolref(use_cte_op.types[i], child_binding);
-                    AddBinding(tempcolref);
+                    pass.temp_column_refs.push_back(make_uniq<BoundColumnRefExpression>(use_cte_op.types[i], child_binding));
+                    AddBinding(*(pass.temp_column_refs.back()));
                     ++count_col_found;
                 }
             }
@@ -586,8 +586,8 @@ void RemoveUnusedColumns::VisitOperator(LogicalOperator &op) {
                     binding = ColumnBinding(aggr.aggregate_index, agg_col);
                 }
 
-                BoundColumnRefExpression tempcolref(cte_op.types[col_idx], binding);
-                cte_side.AddBinding(tempcolref);
+                pass.temp_column_refs.push_back(make_uniq<BoundColumnRefExpression>(cte_op.types[col_idx], binding));
+                cte_side.AddBinding(*(pass.temp_column_refs.back()));
             }
         } else {
             vector<idx_t> cte_root_table_indices = cte_op.GetTableIndex();
@@ -600,8 +600,8 @@ void RemoveUnusedColumns::VisitOperator(LogicalOperator &op) {
             // For each column needed by a CTERef.
             for (auto col_idx : cte_it->second) {
                 ColumnBinding binding(cte_root_table_index, col_idx);
-                BoundColumnRefExpression tempcolref(cte_op.types[col_idx], binding);
-                cte_side.AddBinding(tempcolref);
+                pass.temp_column_refs.push_back(make_uniq<BoundColumnRefExpression>(cte_op.types[col_idx], binding));
+                cte_side.AddBinding(*(pass.temp_column_refs.back()));
             }
         }
         // Now process the duplicated (CTE) side of the CTE.