iClaustron training for Rust developers · module 8 of 9 · 16 min

Advanced scans

Ranges of an ordered index, in order and pruned to a partition; filters run in the data nodes; rows a scan locked, updated and deleted; closing early and errors.

Index scans · beyond module 07's table scans

Index scans: ranges of an ordered index, low and high bounds

ordered index t1_val · its rows in the order of val NULL −299 ⋯ −251 −250 −249 ⋯ −241 −240 −239 ⋯ −201 −200 NULL comes before every value −250 −249 ⋯ −241 low bound −250, in the range low_columns 1 · low_inclusive true high bound −240, not in it high_columns 1 · high_inclusive false

An ordered index keeps its rows in the order of its columns, here val alone (KEY t1_val). A NULL comes before every value.

A range is a low and a high bound, each a row of the key record that says how many leading columns it takes and whether its own values are in.

iclaustron: doc/user-guide.md §5 (Index scans); crates/ic_apid/src/index_scan.rs (ScanRange, encode_range); index_order.rs (NULL first); examples/rust/range_scan.rs (run), lib.rs (EX_VAL_INDEX, val_row)

Index scans · from range_scan.rs

An index scan in Rust: made once, ranges added before each run

let scan = conn.create_index_scan(&index, &val_rec, &t1.rec)?;
// each run: val from -250, included, up to -240, not included
let low = val_row(&t1, &val_rec, -250)?;
let high = val_row(&t1, &val_rec, -240)?;
let range = ScanRange {
  low_columns: 1,
  low_inclusive: true,
  high_columns: 1,
  high_inclusive: false,
  range_no: 0,
};
conn.add_scan_range(scan, &range, &low, &high)?;
let args = ScanArgs { ordered: true, ..ScanArgs::default() };
conn.start_scan(scan, &args)?;
  • Made once: val_rec is the key record, holding every column of the index; t1.rec lays out the rows read.
  • A bound compares as the index orders rows, column by column: on two columns, a low bound (7, 3) takes in (7, 3), (7, 4), (8, 0) …
  • A run reads the ranges added before it, encoded at once; a run given none reads the whole index, NULLs included.
  • Refused with 7163, IC_ERROR_BAD_RANGE: a number above 4 095, more columns than the index has, bounds too long to send.

iclaustron: examples/rust/range_scan.rs (run; the descending pass dropped), lib.rs (val_row); crates/ic_apid/src/scan.rs (create_index_scan, add_scan_range); index_scan.rs (ScanRange, IndexShape::add_range, IndexScan::take_ranges); doc/rust/04-api-design.md (Index scans)

In order · ordered = true

Ordered runs: each fragment's rows merged into the index's order

each fragment's rows in the index's order · one receiver each fragment 0 −250 −247 −244 −241 fragment 1 −249 −246 −243 fragment 2 −248 −245 −242 mergefirst of the heads −250 −249 −248 −247 −246 −245 −244 −243 −242 −241 in val's order, row by row connection: in conn.next_row · executor: in the reader

Each fragment returns its rows in the index's order. An ordered run scans every fragment at once, a receiver each, whatever parallelism says.

The merge gives the first of the receivers' next rows, so it waits for a row or the end from every receiver; it runs on the user thread that reads.

iclaustron: crates/ic_apid/src/scan.rs (parallelism_of, next_merged_row, comes_before); index_order.rs (IndexOrder::key_into); scan_reader.rs (module doc: the reader merges) · the rows' spread over three fragments is illustrative

In order · what an ordered run takes

Ordered runs: read row by row, three collations refused

An ordered run of an index scan
ReadingRow by row only: conn.next_row on a connection, reader.next_row through an executor. A callback, or conn.next_batch, is refused (7127): no batched or asynchronous reading.
Directiondescending: each range from its high end; the ranges keep the order they were added in.
Row recordHolds every column of the index, which the merge compares; else the start is refused (7148, IC_ERROR_KEY_RECORD).
Character columnsCompared in their collation's order, from the weights the data nodes describe when the index is bound.
RefusedAn index with a column in latin2_czech_cs, cp1250_czech_cs or tis620_thai_ci, which weigh a character by its neighbours (7127).

iclaustron: crates/ic_apid/src/scan.rs (ScanArgs: ordered, descending; check_run, next_batch); index_order.rs (IndexOrder::new); collation_map.rs (IC_NEIGHBOUR_COLLATIONS); scan_reader.rs (begin_run); doc/user-guide.md §12

Several ranges · one run, numbered

Several ranges in one run: each row comes with its range's number

range 0 · val = −299 range 1 · −210 to −205 range 2 · −201 to −200 NULL −299 −298 ⋯ −211 −210 ⋯ −205 −204 ⋯ −202 −201 −200 an ordered run gives its rows range by range · range number · val 0 · −299 1 · −210 1 · −209 1 · −208 1 · −207 1 · −206 1 · −205 2 · −201 2 · −200

Three ranges added before one run, numbered in the order added. conn.scan_range_no gives the number of the range the row last read is in.

An ordered run gives its rows range by range, so the numbers must rise in the order added, or the start is refused (7163). An unordered run mixes them.

iclaustron: examples/rust/range_scan.rs (run: ranges; print_run); crates/ic_apid/src/index_scan.rs (RangeSet::note, IndexShape::check_ordered); scan.rs (scan_range_no; comes_before: range number, then order key); crates/ic_port/src/err.rs (IC_ERROR_BAD_RANGE)

Partition pruning · one partition, not all

Partition pruning: a fixed distribution key reads one partition

every range fixes the distribution keyboth bounds the same, all one partition hashed as a key's hintthe low bound's key P0 P1 P2 P3 P4 P5 P6 P7 one partition, one fragment scanned ScanArgs::partition, when set, wins fanout table: ranges fix the basePARTITION_HASH=base:detail:fanout the base's hashdown to a multiple of the fanout P0 P1 P2 P3 P4 P5 P6 P7 the fanout's partitions for that base, here 4 measured, release build: 1 600 scans of one base, 16 at a time · fanout table 28 ms · plain table 99 ms

When every range fixes the distribution key, both bounds the same, and all fall in one partition, the run scans that partition alone.

On a PARTITION_HASH=base:detail:fanout table, ranges that all fix the same base read only the fanout's partitions for that base.

Measured on 100 000 rows in 32 partitions, scans of one base, 100 rows each: the data nodes do an eighth of the work.

iclaustron: crates/ic_apid/src/index_scan.rs (module doc, partition_of_range, fanout_first_of_range); hash.rs (partition_of, fanout_first_of); scan.rs (start_scan: a partition named wins); doc/rust/06-phases.md Phase 10 (fanout_a_scan_by_its_base_skips_the_other_partitions) · 8 partitions drawn

Through an executor · from scan_executor.rs

Index scans through an executor: one shape, a reader per thread

let shape = ScanShape::new_index(&index, &val_rec, &t1.rec)?;

// each thread, read_quarter: its own reader and range
let mut reader = ScanReader::new(&shape);
reader.add_range(&range, &low, &high)?;
let args = ScanArgs {
  ordered: true,
  ..ScanArgs::default()
};
submitter.scan(&mut reader, &args)?;
while reader.next_row()? {
  let val = get_i32(&t1.rec, reader.row(), t1.val).unwrap_or(0);
  // ...
}
  • The shape, an Arc<ScanShape>, is made once and shared; each of the four threads gets its own submitter, a clone, and makes its own reader, with its own range.
  • The executor runs the scan unordered over every fragment; the reader merges an ordered run on its own user thread.
  • reader.range_no gives a row's range number. A slow reader stalls only its own scan.
  • No transaction in the arguments: rows read through a reader cannot be taken over.

iclaustron: examples/rust/scan_executor.rs (run, read_quarter); crates/ic_apid/src/scan_reader.rs (module doc, ScanShape::new_index, ScanReader::add_range, range_no); executor.rs (Submitter::scan; an ordered run unordered here)

Filters · a program run in the data nodes

Scan filters run in the data nodes: only rows that pass travel

API nodea connection, or anexecutor's connectioninputs 40 · 300 · 310 SCAN_REQprogram, inputs data node · on each row the scan reaches id ≥ input 1 AND id < input 2 AND val > input 0 AND name IS NOT NULL 304 · val 40 · name NULL 305 · val 50 · 'row 5' 306 · val 60 · name NULL 307 · val 70 · 'row 7' 308 · val 80 · name NULL 309 · val 90 · 'row 9' 305 · val 50 · 'row 5' 307 · val 70 · 'row 7' 309 · val 90 · 'row 9' passed over passed over passed over ROW_DATA_REP305 · 307 · 309

The program and its inputs go with each run's SCAN_REQ, and run in the data node on every row the scan reaches.

Only rows it passes travel back. A comparison with a NULL is unknown, as in SQL, and only a true condition lets a row through.

iclaustron: examples/rust/conditional_update.rs (write_rows, filter, run part 3: rows 300 to 309, 304 to 309 shown); crates/ic_apid/src/program.rs (module doc: unknown goes with false; ProgramBuilder::compare); doc/rust/05-ndb-protocol.md §9.5

Filters · from conditional_update.rs

Building a scan filter: values, conditions, then inputs per run

let mut b = ProgramBuilder::new(&t1.table);
// ... ours: id from input 1 up to below input 2 ...
let val = b.column(t1.val);
let limit = b.input(0);
let above = b.compare(Cmp::Gt, val, limit);
let name = b.column(t1.name);
let named = b.is_not_null(name);
let wanted = b.and(above, named);
let both = b.and(ours, wanted);
let filter = b.finish(Some(both))?;
let scan = conn.create_scan(&t1.table, &t1.rec)?;
conn.set_scan_program(scan, Some(&filter))?;
conn.set_scan_input(scan, 0, EX_COND_VAL_ABOVE)?;
// ... inputs 1 and 2, then conn.start_scan ...
  • Values: column, constant, integer, input, add, subtract. Conditions: compare (Cmp), is_null, is_not_null, and, or, not.
  • 16 inputs per scan, 0 until set, kept until changed. Through an executor: shape.with_program, reader.set_input.
  • No Result per step: finish returns the first error met in building, and refuses over eight registers at once (7130), different types compared (7164).
  • A filter that assigns is refused (7127).

iclaustron: examples/rust/conditional_update.rs (filter, run part 3); doc/user-guide.md §6; crates/ic_apid/src/program.rs (module doc, ProgramBuilder::finish, IC_PROGRAM_INPUTS, Compiler::eval, same_type, check_program); scan.rs (set_scan_program, set_scan_input); scan_reader.rs (ScanShape::with_program, ScanReader::set_input)

Programs on key operations · for context

Programs on key operations: conditional updates and counters

let mut b = ProgramBuilder::new(&t1.table);
let val = b.column(t1.val);
let step = b.input(0);
let sum = b.add(val, step);
b.assign(t1.val, sum);
let plus = b.finish(None)?;
let counter = conn.create_query(&t1.table, &id_rec, &id_rec)?;
// ... Some(query) from conn.query_mut(counter) ...
query.set_program(Some(&plus))?;
query.set_reads_after(Some(&val_rec))?;
query.set_input(0, 5)?;
// ... an update of row 301, committed ...
let after = conn.query(counter)
  .and_then(|done| get_i32(&val_rec, done.after_row(), t1.val));
  • A counter: the data node adds input 0 to val and reads it back after, in one round trip, the row locked only inside the data node.
  • A condition on a key read, update or delete refuses a row that fails it: 6000 where the operation aborts on error, 626 where it ignores errors.
  • Not on inserts or writes, through a unique index, or on a row taken over from a scan (7127).
  • Through an executor: OpShape::with_program, OpShape::reading_after.

iclaustron: examples/rust/conditional_update.rs (val_plus, run parts 1 and 2); crates/ic_apid/src/query.rs (ApidQuery::set_program, set_reads_after, set_input, after_row); executor.rs (OpShape::with_program, reading_after); transaction.rs (define: where a program may go); doc/user-guide.md §6

Takeover · on a connection

Updating rows a scan finds: a takeover query on its transaction

user thread · connection data nodes SCAN_REQ · exclusive, on the transaction, keys asked a batch · each row locked: ROW_DATA_REP + SCAN_ROW_KEY_REP · SCAN_CONF conn.next_row KEY_OP_REQ · takeover: the row's key, the new name conn.update_scan_row KEY_OP_CONF · in place, the lock now the transaction's SCAN_MORE_REQ · only now: the batch's scan locks go batch handed back after the last batch: COMMIT_REQ, then COMMIT_CONF conn.commit_transaction

An exclusive scan started on a transaction, trans in ScanArgs: each row comes locked, with its key.

conn.update_scan_row defines a query on that transaction: its key from the scan, its values from the query's attribute row.

Its confirmation says the update is in place: the row's lock is now the transaction's, until it commits.

The batch handed back asks for more only once all its takeovers have completed, since asking releases the scan's locks on it.

Commit once the scan is done: one commit for every row the scan updated.

iclaustron: crates/ic_apid/src/scan.rs (take_over_row, last_row_takeover, takeover_settled, queue_more); doc/rust/05-ndb-protocol.md §7.2 (Takeover); doc/user-guide.md §5 (Updating and deleting rows a scan finds)

Takeover · from scan_update.rs

Updating through a scan in Rust: an exclusive scan on a transaction

let trans =
  conn.start_transaction(TransactionHint::Any, Some(&t1.table))?;
let args = ScanArgs {
  lock: ScanLock::Exclusive,
  // The transaction the locks are taken over into.
  trans: Some(trans),
  ..ScanArgs::default()
};
conn.start_scan(scan, &args)?;
while conn.next_row(scan)? {
  // ... qid, the next query, the new name in its attribute row ...
  conn.update_scan_row(qid, scan, &TakeoverArgs::default())?;
}
conn.commit_transaction(trans)?;
  • conn.lock_scan_row takes the lock into the transaction, shared or exclusive as the scan's; delete_scan_row deletes the row.
  • The query is a plain one over the scanned table: through a unique index, with a program or with reads after, it is refused (7127). By default a failed takeover rolls the transaction back.
  • Refused: a scan with no transaction (7127); no row read, or its batch handed back (7162).
  • On a connection only: an executor's scan takes no transaction.

iclaustron: examples/rust/scan_update.rs (run; batch_rows and the query's lookup dropped); crates/ic_apid/src/scan.rs (ScanArgs, TakeoverArgs, lock_scan_row, update_scan_row, delete_scan_row, last_row_takeover); executor.rs (Submitter::scan); transaction.rs (define); doc/user-guide.md §5 (Scans)

No primary key · the hidden key

Tables without a primary key: the scan returns the hidden $PK

let t3 = Counted::bind(&mut conn, "t3")?;
check(
  t3.table.hidden_key() == Some(t3.key),
  "t3's AUTO_INCREMENT column is its hidden key",
)?;
let scan = conn.create_scan(&t3.table, &t3.rec)?;
conn.start_scan(scan, &ScanArgs::default())?;
while conn.next_row(scan)? {
  let row = conn.scan_row(scan);
  // ... get_u64(&t3.rec, row, t3.key) is the row's $PK ...
}
for key in &keys {
  t3.write(&mut conn, key_ops::IC_OP_DELETE, *key, 0)?;
}
  • MySQL adds $PK, BIGINT UNSIGNED, as the table's last column and its key, filled from its AUTO_INCREMENT counter.
  • A record holds it as any column; table.hidden_key() names it, for tools that list columns.
  • An insert takes its value from conn.auto_increment; a scanned row is read, updated or deleted by it later, or taken over at once.

iclaustron: examples/rust/auto_increment.rs (run, Counted::bind, Counted::write: a delete by key); doc/user-guide.md §3, §7; crates/ic_apid/src/auto_increment.rs (TableCounter::of, TableDef::hidden_key, ApidConnection::auto_increment)

Closing early · a scan on a transaction

Closing a locking scan early: its takeovers are confirmed first

Closing early

  • conn.close_scan stops a run at once; batches on their way are dropped.
  • A locking scan whose transaction goes on first waits until its takeovers are confirmed: each keeps its lock.
  • A scan whose transaction rolls back, and any other, stops at once.
  • reader.close waits for the executor, so the reader can run again.

When the run fails

  • The errors are those of every scan (module 07): 296, 274, 266, 4028, 4012, all temporary.
  • Nothing resumes a run. Try again as for any transaction (module 04): a new transaction, the scan and its updates again from the start.

iclaustron: crates/ic_apid/src/scan.rs (close_scan, close_when_settled, stop_waits); scan_reader.rs (ScanReader::close); doc/user-guide.md §5, §9

Not done yet · scans in this release

Scans not done yet: joins, aggregation, events, executor takeovers

  • No pushdown joins, and no aggregation in the data nodes.
  • No events: changes are not streamed to the application.
  • Takeovers on a connection only, not through an executor, whose requests are whole transactions.
  • No program on a row taken over (7127); programs are the basic set, at most eight registers at once.
  • No blob columns: records leave them out, and deleting a row of such a table is refused (7153).

iclaustron: doc/user-guide.md §12; doc/rust/04-api-design.md (Taking scanned rows over: not through an executor); crates/ic_apid/src/transaction.rs (define); scan.rs (delete_scan_row); crates/ic_port/src/err.rs (IC_ERROR_DELETE_WITH_BLOBS)

Recap · module 08

What to take away

  • An index scan is made once; each run reads the ranges added before it, or the whole index.
  • ordered merges every fragment's rows on the user thread, row by row: in next_row or in the reader.
  • Ranges that fix the distribution key read one partition; on a fanout table, the base's partitions.
  • A filter runs in the data nodes: only rows that pass travel, and a NULL comparison passes none.
  • A locking scan on a transaction hands its rows over; a batch's locks go only once its takeovers complete.
  • Scan errors are temporary: start the scan again.

Next · 09 · Inside the crates

iClaustron training for Rust developers · 9 modules · you are at 08 · this module for C developers

Next · 09 · Inside the crates