| 134 | } |
| 135 | |
| 136 | static Status UpdateRows(const shared_ptr<KuduTable>& table, |
| 137 | vector<KuduPredicate*> predicates, int new_val){ |
| 138 | // It's necessary to specify the entire set of key columns when updating a particular row. |
| 139 | // An auto-incrementing column is auto-populated at the server side, and one way to retrieve |
| 140 | // its values is scanning the table with a projection that includes the auto-incrementing column. |
| 141 | KuduScanner scanner(table.get()); |
| 142 | for (int i = 0; i < predicates.size(); i++) { |
| 143 | KUDU_RETURN_NOT_OK(scanner.AddConjunctPredicate(predicates[i])); |
| 144 | } |
| 145 | KUDU_RETURN_NOT_OK(scanner.Open()); |
| 146 | |
| 147 | shared_ptr<KuduSession> session = table->client()->NewSession(); |
| 148 | KUDU_RETURN_NOT_OK(session->SetFlushMode(KuduSession::MANUAL_FLUSH)); |
| 149 | |
| 150 | while (scanner.HasMoreRows()) { |
| 151 | KuduScanBatch batch; |
| 152 | KUDU_RETURN_NOT_OK(scanner.NextBatch(&batch)); |
| 153 | for (KuduScanBatch::const_iterator it = batch.begin(); it != batch.end(); ++it) { |
| 154 | KuduScanBatch::RowPtr row(*it); |
| 155 | int64_t auto_incrementing_counter; |
| 156 | int32_t non_unique_key, val; |
| 157 | |
| 158 | KUDU_RETURN_NOT_OK(row.GetInt32("non_unique_key", &non_unique_key)); |
| 159 | KUDU_RETURN_NOT_OK(row.GetInt64(KuduSchema::GetAutoIncrementingColumnName(), |
| 160 | &auto_incrementing_counter)); |
| 161 | KUDU_RETURN_NOT_OK(row.GetInt32("int_val", &val)); |
| 162 | |
| 163 | unique_ptr<KuduUpdate> new_update(table->NewUpdate()); |
| 164 | KuduPartialRow* update_row = new_update->mutable_row(); |
| 165 | KUDU_RETURN_NOT_OK(update_row->SetInt32("non_unique_key", non_unique_key)); |
| 166 | KUDU_RETURN_NOT_OK(update_row->SetInt64(KuduSchema::GetAutoIncrementingColumnName(), |
| 167 | auto_incrementing_counter)); |
| 168 | KUDU_RETURN_NOT_OK(update_row->SetInt32("int_val", new_val)); |
| 169 | KUDU_RETURN_NOT_OK(session->Apply(new_update.release())); |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | KUDU_RETURN_NOT_OK(session->Flush()); |
| 174 | return session->Close(); |
| 175 | } |
| 176 | |
| 177 | static Status DeleteRows(const shared_ptr<KuduTable>& table, |
| 178 | vector<KuduPredicate*> predicates){ |