Working with Large Databases

Jobs that pump, load or dump databases of many terabytes can run for hours or days. Over that time a server restart, a network drop or a lock in the database is likely. This guide explains how a job reads and writes a large table, how it tracks its progress, how it continues after a failure without loading everything again, and what to set up so that it does.

Most of it works on its own. What you control is:

• the jobs configuration of the server (jobs: section),
• the job options in the UI (Concurrency, Load Options, Determinism, Dictionary),
• the tables themselves (primary keys, database statistics).

How a large table is read

A SQL reader reads a table in one of two ways:

Keyset chunks (default when the table has a usable key)

  SELECT ... WHERE key > last_key ORDER BY key LIMIT 10000   -- chunk 1
  SELECT ... WHERE key > last_key ORDER BY key LIMIT 10000   -- chunk 2
  ...                                                         -- until empty

One cursor

  SELECT ... FROM table                                       -- one long query

• Keyset chunks: the table is read in order of its key, 10,000 rows per query, each query starting after the last key read. Each query is short, so no query holds the database for hours, and the job always knows the position it reached. This is what lets a job resume and read a table in parallel ranges.
• One cursor: one query returns the whole table. It is as fast, but a job that stops in the middle of it has no position to continue from: the table starts over.

The key is the primary key of the table, with columns of number or text types. Unique indexes are not used as a key. Tables without a usable primary key fall back to a physical row position where the database has one:

When a table is read with one cursor

Even with a key, a table is read with one cursor, and cannot resume or be split, when the job:

• reads it with a custom query, or a custom order,
• reads from an offset, or takes the last or random records of a limit,
• reads a PostgreSQL datasource with snapshotRead (one consistent snapshot needs one query),
• synthesizes the entity instead of reading it.

Row counts and progress

The progress bar of an entity compares the records read with its total. To show it without counting a huge table first, the job uses the estimate of the database:

The estimate can be off, so the percentage is approximate. A filtered read (a where or a custom query) can be far smaller than the table, so for those the job also runs an exact COUNT when the table is estimated under 5,000,000 rows. It runs alongside the stream when the driver has a spare connection, and before the stream otherwise. Bigger filtered tables are not counted, because counting them could take as long as reading them: their total stays approximate.

Without statistics there is no estimate, the total falls back to the size of the last scan, and the table may not be split into ranges. Run ANALYZE (PostgreSQL), gather statistics (Oracle) or scan the model again before a large job.

Progress, heartbeats and stalled jobs

A job reports progress as it reads and writes. Long steps that move no records (counting, estimating, splitting a table, taking a snapshot, rebuilding constraints, saving the dictionary) send heartbeats instead, which also appear in the log at most once a minute per step, for example:

estimating orders...
splitting orders...

Two things watch over a running job:

• The stall timer: a job with no progress, no heartbeat and no log line for jobs.stallTimeoutSec seconds (15 minutes by default) is stopped and ends failed with job stalled: no progress for N seconds. It catches a query stuck on a lock or a dead connection.
• The lease: the worker running the job renews it every jobs.heartbeatSec seconds. When the worker dies, the other workers end the job, or resume it with jobs.autoResume. See Interrupted jobs and Leases and heartbeats.

A job counts as completed only when every entity completed: a failed entity, a stall or a fatal error end it failed, even when Continue Streaming On Fail let the other entities finish.

Resuming a failed or interrupted job

Restart on a failed pump, load or dump job runs it again and goes on from where it stopped (see Restart jobs that failed). With jobs.autoResume: true the same happens on its own when the worker of the job dies or the server stops.

Checkpoints

Each time a batch is committed to the sink, the job saves a checkpoint for the entity in MongoDB: the last key (or position) read and the number of records committed. Before each checkpoint, the new values of the dictionary are saved too, so a resumed entity masks the records after the checkpoint with the same values as before.

orders   [=========== committed ===========|-- pending --|            ]
                                           ^
                                 checkpoint: key 48,210,000, 48.2M committed

Restart: the sink has 48.2M rows -> read from key 48,210,001, no truncate

On a restart:

1. Entities that completed are skipped.
2. For each other entity, the job counts the rows of the sink and compares them with the checkpoint. A batch that was being written when the job stopped may or may not have gone in: the job counts the key range of that batch to know.
3. If they match, the entity goes on after the checkpoint, without truncating or dropping the table. The log says resuming orders after record 48210001 (48210000 committed).
4. If not, the entity starts over: it is emptied and read from the beginning. The log says cannot resume orders, it starts over: ... with the reason.

What starts over

An entity starts over instead of resuming when:

A dump to a dataset can skip the entities it finished, but an entity it was writing starts over, because a dataset cannot count its rows.

What does not reset

• The dictionary of the job is kept, not overwritten: values masked in the first run keep their masked value. Without the cache, the job keeps its dictionary in a file in the temporary directory of the host (gig-dict-<jobId>.db) while it can still be restarted, and deletes it when the job ends well or starts afresh.
• The count of automatic resumes (jobs.maxAutoResumes) is kept across restarts. Rerun starts a new job from scratch and resets it.

Reading a big table in parallel ranges

With Concurrency above 1, a job processes several entities at once. A table big enough is also split into ranges of its key, read and written side by side by separate pipelines:

orders (12M rows, Concurrency 4, rangeRows 1,000,000)

  key  1 ......... 3.1M ......... 6.0M ......... 9.2M ......... 12M
       [ range 1  ][   range 2   ][   range 3   ][   range 4    ]
         pipeline     pipeline       pipeline       pipeline

• The number of ranges is the Concurrency of the job (or the most of rangeParts), but no more than one range per 1,000,000 estimated rows (rangeRows). A table needs at least 2,000,000 estimated rows to be split. With shard helpers, a table is split into up to Concurrency × (1 + jobs.shardWorkers) ranges.
• The bounds come from a sample of the key (see the table in How a large table is read), so ranges hold about the same number of rows. The log says reading orders in 4 ranges.
• Ranges share the Concurrency slots with the other entities of the job.
• The first range to write opens and empties the sink table before the others write.
• Each range saves its own checkpoints. A restart resumes every range from its own checkpoint, so the ranges must all have one.
• With a dictionary, ranges that meet the same original value at the same time keep the first masked value saved, so the same value is masked the same way in every range.

jobs.concurrent sets a minimum Concurrency for every job of the server, so you can turn on ranges without editing the options of each rule.

When a table is read whole

A table is read by a single pipeline, even when big, when:

• it is read with one cursor, or by record position (datasets),
• the sink writes with one writer: datasets, file storage, DB2 for i over SSH,
• the sink is DB2 (ibm_db or js400 connections) and the load mode is merge or update, which go through one temporary table,
• it resumes an earlier run that read it in one range,
• it has a limit, a custom query or an offset,
• the job uses Deterministic mode or the entity a seed,
• it is synthesized or learned,
• a transform depends on the order of the records: sequential numbers, values picked from a list, a shuffle, unique values.

If the database cannot sample the key, the log says orders is read in one range, its key could not be split and the job goes on with one pipeline.

Shard helpers for pumps

One job worker is one Node.js process and runs one job at a time. On a pump whose bottleneck is that process (masking is CPU bound), other job workers of the same host can stream ranges of the same job. Set jobs.shardWorkers to the number of helpers per pump, and start enough job workers with gig start -j=N (see start).

pump job (worker 1)        units queue in MongoDB         helpers (pump-shard@host)
  plans the ranges   --->  [orders 1][orders 2][orders 3]  <--- worker 2
  streams units too        [orders 4][customers][...]      <--- worker 3
  sums progress

• Helpers only run on the host of the pump, because they share the dictionary file of the job (SQLite in WAL mode).
• They log into the log of the pump, prefixed [shard worker N], and are hidden from the job lists.
• If a helper dies, its units fail and the pump ends failed: restart it to resume them.
• Helpers are used only for pumps into PostgreSQL, MySQL, Oracle or MongoDB datasources, and not with merge or update load modes.

See Shard helper workers.

Writing to PostgreSQL with COPY

PostgreSQL sinks load each batch with COPY FROM STDIN instead of INSERT statements (jobs.copy, on by default). Merge and update load modes copy into a temporary staging table and apply it in the same transaction. A batch that COPY cannot load is inserted instead; after 3 such batches the table goes on with INSERT. Tables with composite type columns always use INSERT. See COPY writes for Postgres.

Constraints and triggers

Loading a large table with its foreign keys and triggers on is slow and depends on the order of the tables. Each driver turns them off for the tables of the job only, and puts them back at the end:

• PostgreSQL: drops the constraints of the job tables and the foreign keys of other tables that point to them, and saves their definitions in gigantics_meta.dropped_constraints in the sink. If the job dies before restoring them, the next job on that sink restores them. Tables are emptied with TRUNCATE. See PostgreSQL.
• SQL Server: disables the constraints and triggers of the job tables (NOCHECK CONSTRAINT, DISABLE TRIGGER) and unchecks the foreign keys of other tables that point to them, then enables them all again. The datasource option trustConstraints (off by default) re-validates the rows when enabling them, so SQL Server trusts them again, at the cost of a full check. See SQL Server.
• MySQL: empties tables with TRUNCATE with FOREIGN_KEY_CHECKS=0. See MySQL.

Putting constraints back on tables of hundreds of millions of rows can take hours. A failed or canceled job waits for it up to jobs.finishTimeoutSec seconds. See also Oracle and MongoDB.

Job logs and the dictionary

Large jobs write a lot, in two places:

• Job logs are saved in MongoDB. With the default jobs.logTraceLevel: 2, the SQL of each query and the lines of each batch are not saved. Set jobs.logRetentionDays to delete old lines. The log view loads the latest 2,000 lines and more on request.
• The dictionary of a job (when masking with one) is kept in a SQLite file on the host while the job runs, and its new values are saved to MongoDB every 100,000 values or 60 seconds, and before each checkpoint, rather than all at the end, so a resumed job finds the masked values of the batches it committed. Make sure the temporary directory of the host has room for the dictionary of your largest job. See Planning disk space.

Where these options are set

Advanced read options

The rules of Gigantics jobs use the defaults of these options. They can only be changed in the rules: section of a debaser.yml file used with the debaser engine directly (gig debaser):

rules:
  big-pump:
    concurrent: 8
    # auto: count filtered reads under 5M rows; always; never
    exactCount: never
    # auto: keyset chunks when the table has a key; cursor: one query
    readMode: auto
    # rows per keyset query (default 10000)
    readChunkRows: 50000
    # one range per this many estimated rows (default 1000000, 0: never split)
    rangeRows: 500000
    # the most ranges per table (default: concurrent)
    rangeParts: 8

readMode: cursor forces one query per table: those tables cannot resume or be split.

Checklist

Before a large pump, load or dump:

1. Primary keys: make sure the large tables have a primary key with number or text columns. Tables without one are read with one cursor (except on PostgreSQL 14 or later, SQLite and MongoDB) and start over after a failure.
2. Statistics: run ANALYZE (PostgreSQL) or gather statistics (Oracle) so estimates, progress and ranges are right. On PostgreSQL 14 or later, scan the model again if the log asks for it.
3. Concurrency: set Concurrency to 2 or more (or jobs.concurrent) so big tables are read in ranges, within the connections your databases accept.
4. Random mode: on huge tables avoid Deterministic mode, shuffles, limits and custom queries, which turn off resume, ranges or both.
5. Load mode: prefer Truncate. An Append entity that cannot resume fails instead of starting over.
6. Auto resume: set jobs.autoResume: true so restarts and deploys do not end long jobs.
7. Shard helpers: for CPU-bound pumps, set jobs.shardWorkers and start enough job workers (gig start -j=N).
8. Disk and logs: leave room in the temporary directory for the dictionary file, and set jobs.logRetentionDays.
9. Timeouts: raise jobs.stallTimeoutSec if source queries take long before their first rows, and jobs.finishTimeoutSec if rebuilding constraints can take more than a day.

When a job fails, open its log to see why, fix the cause, and use Restart (not Rerun) to continue from the checkpoints. Check the resuming ... and cannot resume ... lines at the start of the log to see which entities continue and which start over.

See Also

• Jobs configuration
• Jobs
• Shuffle
• Planning disk space