From eabab2e2a44b715c914182544e57899323234f86 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Thu, 30 Jul 2020 18:16:52 -0700 Subject: WIP: sorted release ident dumps --- extra/sql_dumps/dump_idents.sql | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/extra/sql_dumps/dump_idents.sql b/extra/sql_dumps/dump_idents.sql index e8126347..d9777ea1 100644 --- a/extra/sql_dumps/dump_idents.sql +++ b/extra/sql_dumps/dump_idents.sql @@ -10,5 +10,21 @@ COPY (SELECT id, rev_id, redirect_id FROM webcapture_ident WHERE is_live=true) COPY (SELECT id, rev_id, redirect_id FROM release_ident WHERE is_live=true) TO '/tmp/fatcat_ident_releases.tsv' WITH NULL ''; COPY (SELECT id, rev_id, redirect_id FROM work_ident WHERE is_live=true) TO '/tmp/fatcat_ident_works.tsv' WITH NULL ''; COPY (SELECT id, editgroup_id, timestamp FROM changelog) TO '/tmp/fatcat_ident_changelog.tsv' WITH NULL ''; +COPY ( + SELECT + release_ident.id, + release_ident.rev_id, + release_ident.redirect_id, + release_rev.work_ident_id + FROM + release_ident + LEFT JOIN release_rev ON release_ident.rev_id = release_rev.id + WHERE + release_ident.is_live=true + AND release_ident.redirect_id IS NULL + AND release_ident.rev_id IS NOT NULL + ORDER BY + release_rev.work_ident_id ASC NULLS LAST +) TO '/tmp/fatcat_ident_releases_by_work.tsv' WITH NULL ''; ROLLBACK; -- cgit v1.2.3 From 902d04b20767d27adc84432f01eab39d008bd8a1 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 4 Aug 2020 18:00:05 -0700 Subject: rust Makefile: fix test command --- rust/Makefile | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/rust/Makefile b/rust/Makefile index 81fb32af..3bfc1b49 100644 --- a/rust/Makefile +++ b/rust/Makefile @@ -10,8 +10,7 @@ help: ## Print info about all commands .PHONY: test test: ## Run all tests and lints - cargo test -- - pipenv run mypy fatcat_scholar/*.py tests/ --ignore-missing-imports + cargo test -- --test-threads 1 .PHONY: dev dev: ## Run web service locally, with reloading -- cgit v1.2.3 From 13e62f51b71473cb30ac4efbd5e1d178d30cde7c Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 4 Aug 2020 18:02:39 -0700 Subject: group-by-work mode for fatcat-export --- rust/src/bin/fatcat-export.rs | 172 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 157 insertions(+), 15 deletions(-) diff --git a/rust/src/bin/fatcat-export.rs b/rust/src/bin/fatcat-export.rs index f52c7d17..3cc0c7a8 100644 --- a/rust/src/bin/fatcat-export.rs +++ b/rust/src/bin/fatcat-export.rs @@ -36,7 +36,8 @@ arg_enum! { Fileset, Webcapture, Release, - Work + Work, + ReleaseByWork, } } @@ -44,6 +45,7 @@ struct IdentRow { ident_id: FatcatId, rev_id: Option, redirect_id: Option, + group_id: Option, } macro_rules! generic_loop_work { @@ -87,6 +89,39 @@ generic_loop_work!(loop_work_webcapture, WebcaptureEntity); generic_loop_work!(loop_work_release, ReleaseEntity); generic_loop_work!(loop_work_work, WorkEntity); +fn loop_work_release_by_work( + row_receiver: channel::Receiver>, + output_sender: channel::Sender>, + db_conn: &DbConn, + expand: Option, +) { + let result: Result<()> = (|| { + for row_batch in row_receiver { + let mut resp_batch: Vec = vec![]; + for row in row_batch { + let mut entity = ReleaseEntity::db_get_rev( + db_conn, + row.rev_id.expect("valid, non-deleted row"), + HideFlags::none(), + )?; // .chain_err(|| "reading entity from database")?; + entity.state = Some("active".to_string()); // only active lines were passed + entity.ident = Some(row.ident_id.to_string()); + if let Some(expand) = expand { + entity.db_expand(db_conn, expand)? + // chain_err(|| "expanding sub-entities from database")?; + } + resp_batch.push(serde_json::to_string(&entity)?); + } + output_sender.send(resp_batch); + } + Ok(()) + })(); + if let Err(ref e) = result { + error!("{}", e); // e.display_chain()) + } + result.unwrap() +} + fn loop_printer( output_receiver: channel::Receiver, done_sender: channel::Sender<()>, @@ -104,11 +139,37 @@ fn loop_printer( Ok(()) } +fn loop_batch_printer( + output_receiver: channel::Receiver>, + done_sender: channel::Sender<()>, +) -> Result<()> { + let output = std::io::stdout(); + // TODO: should log? + // let mut buf_output = BufWriter::new(output.lock()); + let mut buf_output = BufWriter::new(output); + for batch in output_receiver { + for line in batch { + buf_output.write_all(&line.into_bytes())?; + buf_output.write_all(b"\n")?; + buf_output.flush()?; + } + } + drop(done_sender); + Ok(()) +} + fn parse_line(s: &str) -> Result { let fields: Vec = s.split('\t').map(|v| v.to_string()).collect(); - if fields.len() != 3 { + let group_id: Option = if fields.len() == 4 { + match fields[3].as_ref() { + "" => None, + val => Some(FatcatId::from_uuid(&Uuid::from_str(&val)?)), + } + } else if fields.len() == 3 { + None + } else { bail!("Invalid input line"); - } + }; Ok(IdentRow { ident_id: FatcatId::from_uuid(&Uuid::from_str(&fields[0])?), rev_id: match fields[1].as_ref() { @@ -119,6 +180,7 @@ fn parse_line(s: &str) -> Result { "" => None, val => Some(FatcatId::from_uuid(&Uuid::from_str(&val)?)), }, + group_id, }) } @@ -137,6 +199,9 @@ fn test_parse_line() { .is_err() ); assert!(parse_line("00000000-0000-0000-3333-000000000002\t00000000-0000-0000-3333-fff000000001\t00000000-0000-0000-3333-000000000001").is_ok()); + assert!(parse_line("00000000-0000-0000-3333-000000000002\t00000000-0000-0000-3333-fff000000001\t00000000-0000-0000-3333-000000000001\t").is_ok()); + assert!(parse_line("00000000-0000-0000-3333-000000000002\t00000000-0000-0000-3333-fff000000001\t00000000-0000-0000-3333-000000000001\t\t").is_err()); + assert!(parse_line("00000000-0000-0000-3333-000000000002\t00000000-0000-0000-3333-fff000000001\t00000000-0000-0000-3333-000000000001\t00000000-0000-0000-3333-000000000002").is_ok()); } // Use num_cpus/2, or CLI arg for worker count @@ -192,6 +257,7 @@ pub fn do_export( ExportEntityType::Work => { thread::spawn(move || loop_work_work(row_receiver, output_sender, &db_conn, expand)) } + ExportEntityType::ReleaseByWork => unimplemented!(), }; } drop(output_sender); @@ -221,22 +287,97 @@ pub fn do_export( Ok(()) } +pub fn do_export_batch( + num_workers: usize, + expand: Option, + entity_type: ExportEntityType, + redirects: bool, +) -> Result<()> { + let db_pool = server::database_worker_pool()?; + let buf_input = BufReader::new(std::io::stdin()); + let (row_sender, row_receiver) = channel::bounded(CHANNEL_BUFFER_LEN); + let (output_sender, output_receiver) = channel::bounded(CHANNEL_BUFFER_LEN); + let (done_sender, done_receiver) = channel::bounded(0); + + info!("Starting an export of {} entities", entity_type); + + // Start row worker threads + assert!(num_workers > 0); + for _ in 0..num_workers { + let db_conn = db_pool.get().expect("database pool"); + let row_receiver = row_receiver.clone(); + let output_sender = output_sender.clone(); + match entity_type { + ExportEntityType::ReleaseByWork => thread::spawn(move || { + loop_work_release_by_work(row_receiver, output_sender, &db_conn, expand) + }), + _ => unimplemented!(), + }; + } + drop(output_sender); + // Start printer thread + thread::spawn(move || { + loop_batch_printer(output_receiver, done_sender).expect("printing to stdout") + }); + + let mut count = 0; + let mut last_group_id: Option = None; + let mut batch = vec![]; + for line in buf_input.lines() { + let line = line?; + let row = parse_line(&line)?; + match (row.rev_id, row.redirect_id, redirects) { + (None, _, _) => (), + (Some(_), Some(_), false) => (), + _ => { + if row.group_id == None || row.group_id != last_group_id { + if batch.len() > 0 { + row_sender.send(batch); + batch = vec![]; + } + } + last_group_id = row.group_id; + batch.push(row); + } + } + count += 1; + if count % 1000 == 0 { + info!("processed {} lines...", count); + } + } + if batch.len() > 0 { + row_sender.send(batch); + } + drop(row_sender); + done_receiver.recv(); + info!( + "Done reading ({} lines), waiting for workers to exit...", + count + ); + Ok(()) +} + fn main() -> Result<()> { let m = App::new("fatcat-export") .version(env!("CARGO_PKG_VERSION")) .author("Bryan Newbold ") .about("Fast exports of database entities from an id list") - .arg(Arg::from_usage(" 'what entity type the idents correspond to'") - .possible_values(&ExportEntityType::variants()) - .case_insensitive(true)) + .arg( + Arg::from_usage(" 'what entity type the idents correspond to'") + .possible_values(&ExportEntityType::variants()) + .case_insensitive(true), + ) .args_from_usage( "-j --workers=[workers] 'number of threads (database connections) to use' -q --quiet 'less status output to stderr' --include-redirects 'include redirected idents (normally skipped)' - --expand=[expand] 'sub-entities to include in dump'") - .after_help("Reads a ident table TSV dump from stdin (aka, ident_id, rev_id, redirect_id), \ + --expand=[expand] 'sub-entities to include in dump'", + ) + .after_help( + "Reads a ident table TSV dump from stdin (aka, ident_id, rev_id, redirect_id), \ and outputs JSON (one entity per line). Database connection info read from environment \ - (DATABASE_URL, same as fatcatd).") + (DATABASE_URL, same as fatcatd).", + ) .get_matches(); let num_workers: usize = match m.value_of("workers") { @@ -255,10 +396,11 @@ fn main() -> Result<()> { let env = env_logger::Env::default().filter_or(env_logger::DEFAULT_FILTER_ENV, log_level); env_logger::Builder::from_env(env).init(); - do_export( - num_workers, - expand, - value_t_or_exit!(m.value_of("entity_type"), ExportEntityType), - m.is_present("include_redirects"), - ) + let entity_type = value_t_or_exit!(m.value_of("entity_type"), ExportEntityType); + let include_redirects = m.is_present("include_redirects"); + if entity_type == ExportEntityType::ReleaseByWork { + do_export_batch(num_workers, expand, entity_type, include_redirects) + } else { + do_export(num_workers, expand, entity_type, include_redirects) + } } -- cgit v1.2.3 From 8ebd17622af4bb3222560101c447fe1edc9c04c7 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 4 Aug 2020 18:03:06 -0700 Subject: update SQL dump docs with group-by-work command (by default) --- extra/sql_dumps/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extra/sql_dumps/README.md b/extra/sql_dumps/README.md index f24d3d92..1fa37981 100644 --- a/extra/sql_dumps/README.md +++ b/extra/sql_dumps/README.md @@ -30,7 +30,7 @@ Dump locally to stdout, eg: Or, in production: # production, as 'fatcat' user, in /srv/fatcat/src/rust: - cat /tmp/fatcat_ident_releases.tsv | ./target/release/fatcat-export release --expand files,filesets,webcaptures,container -j8 | pigz > /srv/fatcat/snapshots/release_export_expanded.json.gz + cat /tmp/fatcat_ident_releases_by_work.tsv | ./target/release/fatcat-export releasebywork --expand files,filesets,webcaptures,container -j8 | pigz > /srv/fatcat/snapshots/release_export_expanded.json.gz cat /tmp/fatcat_ident_creators.tsv | ./target/release/fatcat-export creator -j8 | pigz > /srv/fatcat/snapshots/creator_export.json.gz cat /tmp/fatcat_ident_containers.tsv | ./target/release/fatcat-export container -j8 | pigz > /srv/fatcat/snapshots/container_export.json.gz cat /tmp/fatcat_ident_files.tsv | ./target/release/fatcat-export file -j8 | pigz > /srv/fatcat/snapshots/file_export.json.gz -- cgit v1.2.3 From 6bc3902b1a421e0b528facce584aad66c0daf7fe Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 4 Aug 2020 18:15:01 -0700 Subject: include releases_by_work in ident tarball --- extra/sql_dumps/ident_table_snapshot.sh | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/extra/sql_dumps/ident_table_snapshot.sh b/extra/sql_dumps/ident_table_snapshot.sh index dbd4caf0..b287a0ce 100755 --- a/extra/sql_dumps/ident_table_snapshot.sh +++ b/extra/sql_dumps/ident_table_snapshot.sh @@ -32,6 +32,7 @@ tar -C /tmp -c --verbose \ fatcat_ident_filesets.tsv \ fatcat_ident_webcaptures.tsv \ fatcat_ident_releases.tsv \ - fatcat_ident_works.tsv + fatcat_ident_works.tsv \ + fatcat_ident_releases_by_work.tsv echo "Done: $OUTFILE" -- cgit v1.2.3 From 4afefd1f2d5cfe71fa7a52ff89aec63af632c874 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 4 Aug 2020 18:21:54 -0700 Subject: proposal for work grouping --- proposals/20200804_grouped_release_exports.md | 60 +++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 proposals/20200804_grouped_release_exports.md diff --git a/proposals/20200804_grouped_release_exports.md b/proposals/20200804_grouped_release_exports.md new file mode 100644 index 00000000..d75ba687 --- /dev/null +++ b/proposals/20200804_grouped_release_exports.md @@ -0,0 +1,60 @@ + +Grouped Release Exports +======================= + +Want to have rich "work" entity dumps, without duplicating with enriched +release dumps. The motivation for this is to do work-level processing of +releases. For example, transform into scholar.archive.org (indexed at work +level), bulk downloading fulltext (only need "best copy for entire work"), +citation analysis, etc. + +One elegant way to do this would be having release exports be sorted by +`work_id`, with all release entities with the same `work_id` contiguous (but +not all on the same line). + +Plan is to update fatcat-export tool (rust) to read in an ident list with +`work_id` included (and sorted by the `work_id`). + +## Rust Implementation + +Remembering that `fatcat-export` operates on lines, and uses channels and +worker threads for processing and serialization, we can't just assume that the +output lines will be in the same order as the input lines (eg, worker threads +can shift order). To preserve order, we will add a new dump mode which operates +on batches (`Vec`) of rows instead, and write the batches contiguously. The +batches themselves may end up out of order, but that is fine. + +## SQL Ident Dump + + +Database timing (running 2020-08-04 on fatcat-qa): + + COPY (SELECT id, rev_id, redirect_id FROM release_ident WHERE is_live=true) TO '/tmp/fatcat_ident_releases.tsv' WITH NULL ''; + => COPY 142635153 + => Time: 143264.483 ms (02:23.264) + + COPY ( + SELECT + release_ident.id, + release_ident.rev_id, + release_ident.redirect_id, + release_rev.work_ident_id + FROM + release_ident + LEFT JOIN release_rev ON release_ident.rev_id = release_rev.id + WHERE + release_ident.is_live=true + AND release_ident.redirect_id IS NULL + AND release_ident.rev_id IS NOT NULL + ORDER BY + release_rev.work_ident_id ASC NULLS LAST + ) TO '/tmp/fatcat_ident_releases_by_work.tsv' WITH NULL ''; + => COPY 142635147 + => Time: 610540.112 ms (10:10.540) + +Much slower, but not a blocking issue. Apparently postgresql will do a full +rowscan instead of using the existing `work_id` sorted index on `release_rev` +when going of a large fraction of the table (here almost the entire table) +because it thinks the read will be so much faster. I don't think this is true +with SSDs? Haven't actually confirmed this is the behavior, just assuming based +on postgresql docs. -- cgit v1.2.3 From b2b830164defc13cb498ba64b1529b4b7f5f1da5 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Wed, 5 Aug 2020 11:11:49 -0700 Subject: fatcat export: flush after batch, not per-line Good catch, thanks Martin --- rust/src/bin/fatcat-export.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/bin/fatcat-export.rs b/rust/src/bin/fatcat-export.rs index 3cc0c7a8..7d671b9a 100644 --- a/rust/src/bin/fatcat-export.rs +++ b/rust/src/bin/fatcat-export.rs @@ -151,8 +151,8 @@ fn loop_batch_printer( for line in batch { buf_output.write_all(&line.into_bytes())?; buf_output.write_all(b"\n")?; - buf_output.flush()?; } + buf_output.flush()?; } drop(done_sender); Ok(()) -- cgit v1.2.3