Partial revert #2972, #2964

Signed-off-by: Andrew Stein <steinlink@gmail.com>

Author: texodus

cpp/perspective/src/cpp/base.cpp

@@ -799,19 +799,6 @@ type_to_dtype<void*>() {
     return DTYPE_OBJECT;
 }
 
-std::ostream&
-operator<<(std::ostream& os, const t_op& op) {
-#define X(NAME)                                                                \
-    case NAME:                                                                 \
-        os << #NAME;                                                           \
-        break;
-
-    switch (op) { FOREACH_T_OP(X) }
-#undef X
-
-    return os;
-}
-
 } // end namespace perspective
 
 namespace std {

cpp/perspective/src/cpp/context_zero.cpp

@@ -10,8 +10,6 @@
 // ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
 // ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
 
-#include "perspective/base.h"
-#include "perspective/raw_types.h"
 #include <perspective/first.h>
 #include <perspective/context_base.h>
 #include <perspective/get_data_extents.h>
@@ -103,7 +101,6 @@ t_ctx0::notify(
         flattened.get_const_column("psp_pkey");
     std::shared_ptr<const t_column> op_sptr =
         flattened.get_const_column("psp_op");
-    auto old_pkey_col = flattened.get_column("psp_old_pkey");
     const t_column* pkey_col = pkey_sptr.get();
     const t_column* op_col = op_sptr.get();
 
@@ -176,20 +173,11 @@ t_ctx0::notify(
             m_symtable.get_interned_tscalar(pkey_col->get_scalar(idx));
         std::uint8_t op_ = *(op_col->get_nth<std::uint8_t>(idx));
         t_op op = static_cast<t_op>(op_);
-        const auto existed = *(existed_col->get_nth<bool>(idx));
-        const auto old_pkey = old_pkey_col->get_scalar(idx);
+        bool existed = *(existed_col->get_nth<bool>(idx));
 
         switch (op) {
             case OP_INSERT: {
-                if (old_pkey.is_valid()) {
-                    m_traversal->move_row(
-                        *m_gstate,
-                        *(m_expression_tables->m_master),
-                        m_config,
-                        old_pkey,
-                        pkey
-                    );
-                } else if (existed) {
+                if (existed) {
                     m_traversal->update_row(
                         *m_gstate,
                         *(m_expression_tables->m_master),

cpp/perspective/src/cpp/data_table.cpp

@@ -23,12 +23,6 @@
 #include <sstream>
 #include <utility>
 namespace perspective {
-std::ostream&
-operator<<(std::ostream& os, const t_flatten_record& fr) {
-    os << "store_idx: " << fr.m_store_idx << ", bidx: " << fr.m_begin_idx
-       << ", eidx: " << fr.m_edge_idx;
-    return os;
-}
 
 void
 t_data_table::set_capacity(t_uindex idx) {
@@ -325,15 +319,15 @@ t_data_table::get_schema() const {
 }
 
 std::shared_ptr<t_data_table>
-t_data_table::flatten(t_uindex limit) const {
+t_data_table::flatten() const {
     PSP_TRACE_SENTINEL();
     PSP_VERBOSE_ASSERT(m_init, "touching uninited object");
     PSP_VERBOSE_ASSERT(is_pkey_table(), "Not a pkeyed table");
     std::shared_ptr<t_data_table> flattened = std::make_shared<t_data_table>(
         "", "", m_schema, DEFAULT_EMPTY_CAPACITY, BACKING_STORE_MEMORY
     );
     flattened->init();
-    flatten_body<std::shared_ptr<t_data_table>>(flattened, limit);
+    flatten_body<std::shared_ptr<t_data_table>>(flattened);
     return flattened;
 }
 
@@ -641,13 +635,6 @@ t_data_table::join(const std::shared_ptr<t_data_table>& other_table) const {
     PSP_VERBOSE_ASSERT(m_init, "touching uninited object");
 
     if (size() != other_table->size()) {
-#if PSP_DEBUG
-        LOG_DEBUG("Joining current table:");
-        pprint();
-        LOG_DEBUG("on this this table:");
-        other_table->pprint();
-#endif
-
         std::stringstream ss;
         ss << "[t_data_table::join] Cannot join two tables of unequal sizes! "
               "Current size: "

cpp/perspective/src/cpp/flat_traversal.cpp

@@ -388,28 +388,6 @@ t_ftrav::delete_row(t_tscalar pkey) {
     ++m_step_deletes;
 }
 
-void
-t_ftrav::move_row(
-    const t_gstate& gstate,
-    const t_data_table& expression_master_table,
-    const t_config& config,
-    t_tscalar old_pkey,
-    t_tscalar new_pkey
-) {
-    auto old_pkiter = m_pkeyidx.find(old_pkey);
-    bool old_pkey_existed = old_pkiter != m_pkeyidx.end();
-    if (!old_pkey_existed) {
-        LOG_DEBUG("Tried to move pkey that doesn't exist: " << old_pkey);
-        return;
-    }
-    LOG_DEBUG("Moving pkey from: " << old_pkey << " to: " << new_pkey);
-
-    (*m_index)[old_pkiter->second].m_deleted = true;
-    t_mselem mselem;
-    fill_sort_elem(gstate, expression_master_table, config, new_pkey, mselem);
-    m_new_elems[new_pkey] = mselem;
-}
-
 std::vector<t_sortspec>
 t_ftrav::get_sort_by() const {
     return m_sortby;

cpp/perspective/src/cpp/gnode.cpp

@@ -10,7 +10,6 @@
 // ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
 // ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
 
-#include "perspective/raw_types.h"
 #include <perspective/first.h>
 #include <perspective/context_unit.h>
 #include <perspective/context_zero.h>
@@ -49,9 +48,7 @@ calc_negate(t_tscalar val) {
     return val.negate();
 }
 
-t_gnode::t_gnode(
-    t_schema input_schema, t_schema output_schema, t_uindex limit
-) :
+t_gnode::t_gnode(t_schema input_schema, t_schema output_schema) :
     m_mode(NODE_PROCESSING_SIMPLE_DATAFLOW)
 #ifdef PSP_PARALLEL_FOR
     ,
@@ -63,7 +60,6 @@ t_gnode::t_gnode(
     m_output_schema(std::move(output_schema)),
     m_init(false),
     m_id(0),
-    m_limit(limit),
     m_last_input_port_id(0),
     m_pool_cleanup([]() {}) {
     PSP_TRACE_SENTINEL();
@@ -89,10 +85,6 @@ t_gnode::t_gnode(
         existed_schema
     };
     m_epoch = std::chrono::high_resolution_clock::now();
-
-    m_input_schema.add_column(
-        "psp_old_pkey", m_input_schema.get_dtype("psp_pkey")
-    );
 }
 
 t_gnode::~t_gnode() {
@@ -105,8 +97,7 @@ void
 t_gnode::init() {
     PSP_TRACE_SENTINEL();
 
-    m_gstate =
-        std::make_shared<t_gstate>(m_input_schema, m_output_schema, m_limit);
+    m_gstate = std::make_shared<t_gstate>(m_input_schema, m_output_schema);
     m_gstate->init();
 
     // Create and store the main input port, which is always port 0. The next
@@ -132,7 +123,7 @@ t_gnode::init() {
 
     for (const auto& iter : m_input_ports) {
         std::shared_ptr<t_port> input_port = iter.second;
-        input_port->get_table()->flatten(m_limit);
+        input_port->get_table()->flatten();
     }
 
     // Initialize expression-related state
@@ -307,22 +298,16 @@ t_gnode::_process_table(t_uindex port_id) {
     }
 
     m_was_updated = true;
-    flattened = input_port->get_table()->flatten(m_limit);
+    flattened = input_port->get_table()->flatten();
 
     PSP_GNODE_VERIFY_TABLE(flattened);
     PSP_GNODE_VERIFY_TABLE(get_table());
 
     t_uindex flattened_num_rows = flattened->num_rows();
+
     std::vector<t_rlookup> row_lookup(flattened_num_rows);
     t_column* pkey_col = flattened->get_column("psp_pkey").get();
 
-#if PSP_DEBUG
-    LOG_DEBUG("m_mapping");
-    for (const auto [k, v] : m_gstate->get_pkey_map()) {
-        LOG_DEBUG("KEY: " << k << " , VALUE: " << v);
-    }
-#endif
-
     for (t_uindex idx = 0; idx < flattened_num_rows; ++idx) {
         // See if each primary key in flattened already exist in the dataset
         t_tscalar pkey = pkey_col->get_scalar(idx);