Page MenuHomeSoftware Heritage
Differential D7060

preparation refactoring for following diffs
ClosedPublic
Actions

Authored by douardda on Feb 2 2022, 1:49 PM.

Details

Reviewers
vlorentz
Group Reviewers
Reviewers
Maniphest Tasks
T3894: Add a 'swh db upgrade' cli
Commits
rDCORE459da96c1e6b: Move utility functions from cli.db to db.db_utils
rDCORE932eed4863ba: Remove the old and unused log-schema.sql file
Summary

Remove the old and unused log-schema.sql file

Move utility functions from cli.db to db.db_utils

Related to T3894.

Diff Detail

Repository
rDCORE Foundations and core functionalities
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

douardda created this revision.Feb 2 2022, 1:49 PM
Herald added a reviewer: Reviewers. · View Herald TranscriptFeb 2 2022, 1:49 PM
douardda added a child revision: D7061: Make `swh db init` retrieve the db cnx uri from the config file.Feb 2 2022, 1:49 PM
swh-public-ci added a comment.Feb 2 2022, 1:50 PM

Build has FAILED

Patch application report for D7060 (id=25610)

Rebasing onto aba5c80765...

Current branch diff-target is up to date.
Changes applied before test
commit 6c665e38e6579a02839eef6a8a6b60dc332be51c
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 14:49:41 2022 +0100

    Move utility functions from cli.db to db.db_utils

commit e4117fcf51a256c5b708f909c818f452c794c0f8
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 10:07:35 2022 +0100

    Remove the old and unused log-schema.sql file

Link to build: https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/265/
See console output for more information: https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/265/console

Harbormaster returned this revision to the author for changes because remote builds failed.Feb 2 2022, 1:50 PM
Harbormaster failed remote builds in B26440: Diff 25610!
douardda updated this revision to Diff 25617.Feb 2 2022, 2:34 PM

fix

swh-public-ci added a comment.Feb 2 2022, 2:35 PM

Build is green

Patch application report for D7060 (id=25617)

Rebasing onto aba5c80765...

Current branch diff-target is up to date.
Changes applied before test
commit d3b7b081fd3eda460d11d1d6d70f2b52fa3e0ef3
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 14:49:41 2022 +0100

    Move utility functions from cli.db to db.db_utils

commit e4117fcf51a256c5b708f909c818f452c794c0f8
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 10:07:35 2022 +0100

    Remove the old and unused log-schema.sql file

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/270/ for more details.

Harbormaster completed remote builds in B26447: Diff 25617.Feb 2 2022, 2:35 PM
douardda requested review of this revision.Feb 2 2022, 2:35 PM
vlorentz accepted this revision.Feb 2 2022, 3:06 PM
vlorentz added a subscriber: vlorentz.

I don't see why, but ok

swh/core/db/db_utils.py
345 ↗(On Diff #25617)

can make this toplevel now

379 ↗(On Diff #25617)

ditto

This revision is now accepted and ready to land.Feb 2 2022, 3:06 PM
douardda added inline comments.Feb 2 2022, 3:25 PM
swh/core/db/db_utils.py
345 ↗(On Diff #25617)

good catch

douardda updated this revision to Diff 25632.Feb 2 2022, 3:30 PM

move (forgotten) import stmts top level (thx vlorentz)

swh-public-ci added a comment.Feb 2 2022, 3:31 PM

Build is green

Patch application report for D7060 (id=25632)

Rebasing onto aba5c80765...

Current branch diff-target is up to date.
Changes applied before test
commit de1c894e3fa88b3e1e2ba1e7c7dbfc77f7ea4b51
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 14:49:41 2022 +0100

    Move utility functions from cli.db to db.db_utils

commit e4117fcf51a256c5b708f909c818f452c794c0f8
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 10:07:35 2022 +0100

    Remove the old and unused log-schema.sql file

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/285/ for more details.

Harbormaster completed remote builds in B26462: Diff 25632.Feb 2 2022, 3:31 PM
douardda updated this revision to Diff 25928.Feb 11 2022, 12:19 PM

rebase

swh-public-ci added a comment.Feb 11 2022, 12:21 PM

Build is green

Patch application report for D7060 (id=25928)

Rebasing onto 5046a95f05...

Current branch diff-target is up to date.
Changes applied before test
commit 459da96c1e6b6044deb874e0e668606c3da3e43c
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 14:49:41 2022 +0100

    Move utility functions from cli.db to db.db_utils

commit 932eed4863baedf011b4a59d0227331d37db1dfd
Author: David Douard <david.douard@sdfa3.org>
Date:   Fri Jan 28 10:07:35 2022 +0100

    Remove the old and unused log-schema.sql file

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/295/ for more details.

Harbormaster completed remote builds in B26805: Diff 25928.Feb 11 2022, 12:21 PM
Closed by commit rDCORE932eed4863ba: Remove the old and unused log-schema.sql file (authored by douardda). · Explain WhyFeb 11 2022, 12:52 PM
This revision was automatically updated to reflect the committed changes.
douardda added a commit: rDCORE932eed4863ba: Remove the old and unused log-schema.sql file.
douardda added a commit: rDCORE459da96c1e6b: Move utility functions from cli.db to db.db_utils.

Revision Contents
Changeset List

PathSize
1 line
swh/
core/
sql/

Diff 25937

View Options

MANIFEST.in

# Copyright (C) 2019-2020 The Software Heritage developers # Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from swh.storage import get_storage from swh.storage import get_storage
def get_storage_with_buffer_config(**buffer_config): def get_storage_with_buffer_config(**buffer_config):
storage_config = { storage_config = {
"cls": "pipeline", "cls": "pipeline",
"steps": [ "steps": [{"cls": "buffer", **buffer_config}, {"cls": "memory"},],
{"cls": "validate"},
{"cls": "buffer", **buffer_config},
{"cls": "memory"},
],
} }
return get_storage(**storage_config) return get_storage(**storage_config)
def test_buffering_proxy_storage_content_threshold_not_hit(sample_data): def test_buffering_proxy_storage_content_threshold_not_hit(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
contents_dict = [c.to_dict() for c in contents]
storage = get_storage_with_buffer_config(min_batch_size={"content": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 10,})
s = storage.content_add([contents[0], contents[1]]) s = storage.content_add([contents[0], contents[1]])
assert s == {} assert s == {}
# contents have not been written to storage # contents have not been written to storage
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing(contents_dict)
assert set(missing_contents) == set([contents[0]["sha1"], contents[1]["sha1"]]) assert set(missing_contents) == set([contents[0].sha1, contents[1].sha1])
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"content:add": 1 + 1, "content:add": 1 + 1,
"content:add:bytes": contents[0]["length"] + contents[1]["length"], "content:add:bytes": contents[0].length + contents[1].length,
} }
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing(contents_dict)
assert list(missing_contents) == [] assert list(missing_contents) == []
def test_buffering_proxy_storage_content_threshold_nb_hit(sample_data): def test_buffering_proxy_storage_content_threshold_nb_hit(sample_data_model):
contents = sample_data["content"] content = sample_data_model["content"][0]
content_dict = content.to_dict()
storage = get_storage_with_buffer_config(min_batch_size={"content": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 1,})
s = storage.content_add([contents[0]]) s = storage.content_add([content])
assert s == { assert s == {
"content:add": 1, "content:add": 1,
"content:add:bytes": contents[0]["length"], "content:add:bytes": content.length,
} }
missing_contents = storage.content_missing([contents[0]]) missing_contents = storage.content_missing([content_dict])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_content_deduplicate(sample_data): def test_buffering_proxy_storage_content_deduplicate(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
storage = get_storage_with_buffer_config(min_batch_size={"content": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 2,})
s = storage.content_add([contents[0], contents[0]]) s = storage.content_add([contents[0], contents[0]])
assert s == {} assert s == {}
s = storage.content_add([contents[0]]) s = storage.content_add([contents[0]])
assert s == {} assert s == {}
s = storage.content_add([contents[1]]) s = storage.content_add([contents[1]])
assert s == { assert s == {
"content:add": 1 + 1, "content:add": 1 + 1,
"content:add:bytes": contents[0]["length"] + contents[1]["length"], "content:add:bytes": contents[0].length + contents[1].length,
} }
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing([c.to_dict() for c in contents])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_content_threshold_bytes_hit(sample_data): def test_buffering_proxy_storage_content_threshold_bytes_hit(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
content_bytes_min_batch_size = 2 content_bytes_min_batch_size = 2
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"content": 10, "content_bytes": content_bytes_min_batch_size,} min_batch_size={"content": 10, "content_bytes": content_bytes_min_batch_size,}
) )
assert contents[0]["length"] > content_bytes_min_batch_size assert contents[0].length > content_bytes_min_batch_size
s = storage.content_add([contents[0]]) s = storage.content_add([contents[0]])
assert s == { assert s == {
"content:add": 1, "content:add": 1,
"content:add:bytes": contents[0]["length"], "content:add:bytes": contents[0].length,
} }
missing_contents = storage.content_missing([contents[0]]) missing_contents = storage.content_missing([contents[0].to_dict()])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_skipped_content_threshold_not_hit(sample_data): def test_buffering_proxy_storage_skipped_content_threshold_not_hit(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"]
contents_dict = [c.to_dict() for c in contents]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 10,})
s = storage.skipped_content_add([contents[0], contents[1]]) s = storage.skipped_content_add([contents[0], contents[1]])
assert s == {} assert s == {}
# contents have not been written to storage # contents have not been written to storage
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing(contents_dict)
assert {c["sha1"] for c in missing_contents} == {c["sha1"] for c in contents} assert {c["sha1"] for c in missing_contents} == {c.sha1 for c in contents}
s = storage.flush() s = storage.flush()
assert s == {"skipped_content:add": 1 + 1} assert s == {"skipped_content:add": 1 + 1}
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing(contents_dict)
assert list(missing_contents) == [] assert list(missing_contents) == []
def test_buffering_proxy_storage_skipped_content_threshold_nb_hit(sample_data): def test_buffering_proxy_storage_skipped_content_threshold_nb_hit(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 1,})
s = storage.skipped_content_add([contents[0]]) s = storage.skipped_content_add([contents[0]])
assert s == {"skipped_content:add": 1} assert s == {"skipped_content:add": 1}
missing_contents = storage.skipped_content_missing([contents[0]]) missing_contents = storage.skipped_content_missing([contents[0].to_dict()])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_skipped_content_deduplicate(sample_data): def test_buffering_proxy_storage_skipped_content_deduplicate(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 2,})
s = storage.skipped_content_add([contents[0], contents[0]]) s = storage.skipped_content_add([contents[0], contents[0]])
assert s == {} assert s == {}
s = storage.skipped_content_add([contents[0]]) s = storage.skipped_content_add([contents[0]])
assert s == {} assert s == {}
s = storage.skipped_content_add([contents[1]]) s = storage.skipped_content_add([contents[1]])
assert s == { assert s == {
"skipped_content:add": 1 + 1, "skipped_content:add": 1 + 1,
} }
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing([c.to_dict() for c in contents])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_directory_threshold_not_hit(sample_data): def test_buffering_proxy_storage_directory_threshold_not_hit(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 10,})
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == {} assert s == {}
directory_id = directories[0]["id"] directory_id = directories[0].id
missing_directories = storage.directory_missing([directory_id]) missing_directories = storage.directory_missing([directory_id])
assert list(missing_directories) == [directory_id] assert list(missing_directories) == [directory_id]
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"directory:add": 1, "directory:add": 1,
} }
missing_directories = storage.directory_missing([directory_id]) missing_directories = storage.directory_missing([directory_id])
assert list(missing_directories) == [] assert list(missing_directories) == []
def test_buffering_proxy_storage_directory_threshold_hit(sample_data): def test_buffering_proxy_storage_directory_threshold_hit(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 1,})
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == { assert s == {
"directory:add": 1, "directory:add": 1,
} }
missing_directories = storage.directory_missing([directories[0]["id"]]) missing_directories = storage.directory_missing([directories[0].id])
assert list(missing_directories) == [] assert list(missing_directories) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_directory_deduplicate(sample_data): def test_buffering_proxy_storage_directory_deduplicate(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 2,})
s = storage.directory_add([directories[0], directories[0]]) s = storage.directory_add([directories[0], directories[0]])
assert s == {} assert s == {}
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == {} assert s == {}
s = storage.directory_add([directories[1]]) s = storage.directory_add([directories[1]])
assert s == { assert s == {
"directory:add": 1 + 1, "directory:add": 1 + 1,
} }
missing_directories = storage.directory_missing( missing_directories = storage.directory_missing([d.id for d in directories])
[directories[0]["id"], directories[1]["id"]]
)
assert list(missing_directories) == [] assert list(missing_directories) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_revision_threshold_not_hit(sample_data): def test_buffering_proxy_storage_revision_threshold_not_hit(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 10,})
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == {} assert s == {}
revision_id = revisions[0]["id"] revision_id = revisions[0].id
missing_revisions = storage.revision_missing([revision_id]) missing_revisions = storage.revision_missing([revision_id])
assert list(missing_revisions) == [revision_id] assert list(missing_revisions) == [revision_id]
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"revision:add": 1, "revision:add": 1,
} }
missing_revisions = storage.revision_missing([revision_id]) missing_revisions = storage.revision_missing([revision_id])
assert list(missing_revisions) == [] assert list(missing_revisions) == []
def test_buffering_proxy_storage_revision_threshold_hit(sample_data): def test_buffering_proxy_storage_revision_threshold_hit(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 1,})
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == { assert s == {
"revision:add": 1, "revision:add": 1,
} }
missing_revisions = storage.revision_missing([revisions[0]["id"]]) missing_revisions = storage.revision_missing([revisions[0].id])
assert list(missing_revisions) == [] assert list(missing_revisions) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_revision_deduplicate(sample_data): def test_buffering_proxy_storage_revision_deduplicate(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 2,})
s = storage.revision_add([revisions[0], revisions[0]]) s = storage.revision_add([revisions[0], revisions[0]])
assert s == {} assert s == {}
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == {} assert s == {}
s = storage.revision_add([revisions[1]]) s = storage.revision_add([revisions[1]])
assert s == { assert s == {
"revision:add": 1 + 1, "revision:add": 1 + 1,
} }
missing_revisions = storage.revision_missing( missing_revisions = storage.revision_missing([r.id for r in revisions])
[revisions[0]["id"], revisions[1]["id"]]
)
assert list(missing_revisions) == [] assert list(missing_revisions) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_release_threshold_not_hit(sample_data): def test_buffering_proxy_storage_release_threshold_not_hit(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"]
threshold = 10 threshold = 10
assert len(releases) < threshold assert len(releases) < threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"release": threshold,} # configuration set min_batch_size={"release": threshold,} # configuration set
) )
s = storage.release_add(releases) s = storage.release_add(releases)
assert s == {} assert s == {}
release_ids = [r["id"] for r in releases] release_ids = [r.id for r in releases]
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == release_ids assert list(missing_releases) == release_ids
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"release:add": len(releases), "release:add": len(releases),
} }
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == [] assert list(missing_releases) == []
def test_buffering_proxy_storage_release_threshold_hit(sample_data): def test_buffering_proxy_storage_release_threshold_hit(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"]
threshold = 2 threshold = 2
assert len(releases) > threshold assert len(releases) > threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"release": threshold,} # configuration set min_batch_size={"release": threshold,} # configuration set
) )
s = storage.release_add(releases) s = storage.release_add(releases)
assert s == { assert s == {
"release:add": len(releases), "release:add": len(releases),
} }
release_ids = [r["id"] for r in releases] release_ids = [r.id for r in releases]
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == [] assert list(missing_releases) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_release_deduplicate(sample_data): def test_buffering_proxy_storage_release_deduplicate(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"release": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"release": 2,})
s = storage.release_add([releases[0], releases[0]]) s = storage.release_add([releases[0], releases[0]])
assert s == {} assert s == {}
s = storage.release_add([releases[0]]) s = storage.release_add([releases[0]])
assert s == {} assert s == {}
s = storage.release_add([releases[1]]) s = storage.release_add([releases[1]])
assert s == { assert s == {
"release:add": 1 + 1, "release:add": 1 + 1,
} }
missing_releases = storage.release_missing([releases[0]["id"], releases[1]["id"]]) missing_releases = storage.release_missing([r.id for r in releases])
assert list(missing_releases) == [] assert list(missing_releases) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_clear(sample_data): def test_buffering_proxy_storage_clear(sample_data_model):
"""Clear operation on buffer """Clear operation on buffer
""" """
threshold = 10 threshold = 10
contents = sample_data["content"] contents = sample_data_model["content"]
assert 0 < len(contents) < threshold assert 0 < len(contents) < threshold
skipped_contents = sample_data["skipped_content"] skipped_contents = sample_data_model["skipped_content"]
assert 0 < len(skipped_contents) < threshold assert 0 < len(skipped_contents) < threshold
directories = sample_data["directory"] directories = sample_data_model["directory"]
assert 0 < len(directories) < threshold assert 0 < len(directories) < threshold
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
assert 0 < len(revisions) < threshold assert 0 < len(revisions) < threshold
releases = sample_data["release"] releases = sample_data_model["release"]
assert 0 < len(releases) < threshold assert 0 < len(releases) < threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={ min_batch_size={
"content": threshold, "content": threshold,
"skipped_content": threshold, "skipped_content": threshold,
"directory": threshold, "directory": threshold,
"revision": threshold, "revision": threshold,
▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines
View Options

swh/core/sql/log-schema.sql

# Copyright (C) 2019-2020 The Software Heritage developers # Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from swh.storage import get_storage from swh.storage import get_storage
def get_storage_with_buffer_config(**buffer_config): def get_storage_with_buffer_config(**buffer_config):
storage_config = { storage_config = {
"cls": "pipeline", "cls": "pipeline",
"steps": [ "steps": [{"cls": "buffer", **buffer_config}, {"cls": "memory"},],
{"cls": "validate"},
{"cls": "buffer", **buffer_config},
{"cls": "memory"},
],
} }
return get_storage(**storage_config) return get_storage(**storage_config)
def test_buffering_proxy_storage_content_threshold_not_hit(sample_data): def test_buffering_proxy_storage_content_threshold_not_hit(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
contents_dict = [c.to_dict() for c in contents]
storage = get_storage_with_buffer_config(min_batch_size={"content": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 10,})
s = storage.content_add([contents[0], contents[1]]) s = storage.content_add([contents[0], contents[1]])
assert s == {} assert s == {}
# contents have not been written to storage # contents have not been written to storage
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing(contents_dict)
assert set(missing_contents) == set([contents[0]["sha1"], contents[1]["sha1"]]) assert set(missing_contents) == set([contents[0].sha1, contents[1].sha1])
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"content:add": 1 + 1, "content:add": 1 + 1,
"content:add:bytes": contents[0]["length"] + contents[1]["length"], "content:add:bytes": contents[0].length + contents[1].length,
} }
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing(contents_dict)
assert list(missing_contents) == [] assert list(missing_contents) == []
def test_buffering_proxy_storage_content_threshold_nb_hit(sample_data): def test_buffering_proxy_storage_content_threshold_nb_hit(sample_data_model):
contents = sample_data["content"] content = sample_data_model["content"][0]
content_dict = content.to_dict()
storage = get_storage_with_buffer_config(min_batch_size={"content": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 1,})
s = storage.content_add([contents[0]]) s = storage.content_add([content])
assert s == { assert s == {
"content:add": 1, "content:add": 1,
"content:add:bytes": contents[0]["length"], "content:add:bytes": content.length,
} }
missing_contents = storage.content_missing([contents[0]]) missing_contents = storage.content_missing([content_dict])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_content_deduplicate(sample_data): def test_buffering_proxy_storage_content_deduplicate(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
storage = get_storage_with_buffer_config(min_batch_size={"content": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"content": 2,})
s = storage.content_add([contents[0], contents[0]]) s = storage.content_add([contents[0], contents[0]])
assert s == {} assert s == {}
s = storage.content_add([contents[0]]) s = storage.content_add([contents[0]])
assert s == {} assert s == {}
s = storage.content_add([contents[1]]) s = storage.content_add([contents[1]])
assert s == { assert s == {
"content:add": 1 + 1, "content:add": 1 + 1,
"content:add:bytes": contents[0]["length"] + contents[1]["length"], "content:add:bytes": contents[0].length + contents[1].length,
} }
missing_contents = storage.content_missing([contents[0], contents[1]]) missing_contents = storage.content_missing([c.to_dict() for c in contents])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_content_threshold_bytes_hit(sample_data): def test_buffering_proxy_storage_content_threshold_bytes_hit(sample_data_model):
contents = sample_data["content"] contents = sample_data_model["content"]
content_bytes_min_batch_size = 2 content_bytes_min_batch_size = 2
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"content": 10, "content_bytes": content_bytes_min_batch_size,} min_batch_size={"content": 10, "content_bytes": content_bytes_min_batch_size,}
) )
assert contents[0]["length"] > content_bytes_min_batch_size assert contents[0].length > content_bytes_min_batch_size
s = storage.content_add([contents[0]]) s = storage.content_add([contents[0]])
assert s == { assert s == {
"content:add": 1, "content:add": 1,
"content:add:bytes": contents[0]["length"], "content:add:bytes": contents[0].length,
} }
missing_contents = storage.content_missing([contents[0]]) missing_contents = storage.content_missing([contents[0].to_dict()])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_skipped_content_threshold_not_hit(sample_data): def test_buffering_proxy_storage_skipped_content_threshold_not_hit(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"]
contents_dict = [c.to_dict() for c in contents]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 10,})
s = storage.skipped_content_add([contents[0], contents[1]]) s = storage.skipped_content_add([contents[0], contents[1]])
assert s == {} assert s == {}
# contents have not been written to storage # contents have not been written to storage
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing(contents_dict)
assert {c["sha1"] for c in missing_contents} == {c["sha1"] for c in contents} assert {c["sha1"] for c in missing_contents} == {c.sha1 for c in contents}
s = storage.flush() s = storage.flush()
assert s == {"skipped_content:add": 1 + 1} assert s == {"skipped_content:add": 1 + 1}
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing(contents_dict)
assert list(missing_contents) == [] assert list(missing_contents) == []
def test_buffering_proxy_storage_skipped_content_threshold_nb_hit(sample_data): def test_buffering_proxy_storage_skipped_content_threshold_nb_hit(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 1,})
s = storage.skipped_content_add([contents[0]]) s = storage.skipped_content_add([contents[0]])
assert s == {"skipped_content:add": 1} assert s == {"skipped_content:add": 1}
missing_contents = storage.skipped_content_missing([contents[0]]) missing_contents = storage.skipped_content_missing([contents[0].to_dict()])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_skipped_content_deduplicate(sample_data): def test_buffering_proxy_storage_skipped_content_deduplicate(sample_data_model):
contents = sample_data["skipped_content"] contents = sample_data_model["skipped_content"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"skipped_content": 2,})
s = storage.skipped_content_add([contents[0], contents[0]]) s = storage.skipped_content_add([contents[0], contents[0]])
assert s == {} assert s == {}
s = storage.skipped_content_add([contents[0]]) s = storage.skipped_content_add([contents[0]])
assert s == {} assert s == {}
s = storage.skipped_content_add([contents[1]]) s = storage.skipped_content_add([contents[1]])
assert s == { assert s == {
"skipped_content:add": 1 + 1, "skipped_content:add": 1 + 1,
} }
missing_contents = storage.skipped_content_missing([contents[0], contents[1]]) missing_contents = storage.skipped_content_missing([c.to_dict() for c in contents])
assert list(missing_contents) == [] assert list(missing_contents) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_directory_threshold_not_hit(sample_data): def test_buffering_proxy_storage_directory_threshold_not_hit(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 10,})
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == {} assert s == {}
directory_id = directories[0]["id"] directory_id = directories[0].id
missing_directories = storage.directory_missing([directory_id]) missing_directories = storage.directory_missing([directory_id])
assert list(missing_directories) == [directory_id] assert list(missing_directories) == [directory_id]
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"directory:add": 1, "directory:add": 1,
} }
missing_directories = storage.directory_missing([directory_id]) missing_directories = storage.directory_missing([directory_id])
assert list(missing_directories) == [] assert list(missing_directories) == []
def test_buffering_proxy_storage_directory_threshold_hit(sample_data): def test_buffering_proxy_storage_directory_threshold_hit(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 1,})
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == { assert s == {
"directory:add": 1, "directory:add": 1,
} }
missing_directories = storage.directory_missing([directories[0]["id"]]) missing_directories = storage.directory_missing([directories[0].id])
assert list(missing_directories) == [] assert list(missing_directories) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_directory_deduplicate(sample_data): def test_buffering_proxy_storage_directory_deduplicate(sample_data_model):
directories = sample_data["directory"] directories = sample_data_model["directory"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"directory": 2,})
s = storage.directory_add([directories[0], directories[0]]) s = storage.directory_add([directories[0], directories[0]])
assert s == {} assert s == {}
s = storage.directory_add([directories[0]]) s = storage.directory_add([directories[0]])
assert s == {} assert s == {}
s = storage.directory_add([directories[1]]) s = storage.directory_add([directories[1]])
assert s == { assert s == {
"directory:add": 1 + 1, "directory:add": 1 + 1,
} }
missing_directories = storage.directory_missing( missing_directories = storage.directory_missing([d.id for d in directories])
[directories[0]["id"], directories[1]["id"]]
)
assert list(missing_directories) == [] assert list(missing_directories) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_revision_threshold_not_hit(sample_data): def test_buffering_proxy_storage_revision_threshold_not_hit(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 10,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 10,})
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == {} assert s == {}
revision_id = revisions[0]["id"] revision_id = revisions[0].id
missing_revisions = storage.revision_missing([revision_id]) missing_revisions = storage.revision_missing([revision_id])
assert list(missing_revisions) == [revision_id] assert list(missing_revisions) == [revision_id]
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"revision:add": 1, "revision:add": 1,
} }
missing_revisions = storage.revision_missing([revision_id]) missing_revisions = storage.revision_missing([revision_id])
assert list(missing_revisions) == [] assert list(missing_revisions) == []
def test_buffering_proxy_storage_revision_threshold_hit(sample_data): def test_buffering_proxy_storage_revision_threshold_hit(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 1,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 1,})
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == { assert s == {
"revision:add": 1, "revision:add": 1,
} }
missing_revisions = storage.revision_missing([revisions[0]["id"]]) missing_revisions = storage.revision_missing([revisions[0].id])
assert list(missing_revisions) == [] assert list(missing_revisions) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_revision_deduplicate(sample_data): def test_buffering_proxy_storage_revision_deduplicate(sample_data_model):
revisions = sample_data["revision"] revisions = sample_data_model["revision"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"revision": 2,})
s = storage.revision_add([revisions[0], revisions[0]]) s = storage.revision_add([revisions[0], revisions[0]])
assert s == {} assert s == {}
s = storage.revision_add([revisions[0]]) s = storage.revision_add([revisions[0]])
assert s == {} assert s == {}
s = storage.revision_add([revisions[1]]) s = storage.revision_add([revisions[1]])
assert s == { assert s == {
"revision:add": 1 + 1, "revision:add": 1 + 1,
} }
missing_revisions = storage.revision_missing( missing_revisions = storage.revision_missing([r.id for r in revisions])
[revisions[0]["id"], revisions[1]["id"]]
)
assert list(missing_revisions) == [] assert list(missing_revisions) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_release_threshold_not_hit(sample_data): def test_buffering_proxy_storage_release_threshold_not_hit(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"]
threshold = 10 threshold = 10
assert len(releases) < threshold assert len(releases) < threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"release": threshold,} # configuration set min_batch_size={"release": threshold,} # configuration set
) )
s = storage.release_add(releases) s = storage.release_add(releases)
assert s == {} assert s == {}
release_ids = [r["id"] for r in releases] release_ids = [r.id for r in releases]
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == release_ids assert list(missing_releases) == release_ids
s = storage.flush() s = storage.flush()
assert s == { assert s == {
"release:add": len(releases), "release:add": len(releases),
} }
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == [] assert list(missing_releases) == []
def test_buffering_proxy_storage_release_threshold_hit(sample_data): def test_buffering_proxy_storage_release_threshold_hit(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"]
threshold = 2 threshold = 2
assert len(releases) > threshold assert len(releases) > threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={"release": threshold,} # configuration set min_batch_size={"release": threshold,} # configuration set
) )
s = storage.release_add(releases) s = storage.release_add(releases)
assert s == { assert s == {
"release:add": len(releases), "release:add": len(releases),
} }
release_ids = [r["id"] for r in releases] release_ids = [r.id for r in releases]
missing_releases = storage.release_missing(release_ids) missing_releases = storage.release_missing(release_ids)
assert list(missing_releases) == [] assert list(missing_releases) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_release_deduplicate(sample_data): def test_buffering_proxy_storage_release_deduplicate(sample_data_model):
releases = sample_data["release"] releases = sample_data_model["release"][:2]
storage = get_storage_with_buffer_config(min_batch_size={"release": 2,}) storage = get_storage_with_buffer_config(min_batch_size={"release": 2,})
s = storage.release_add([releases[0], releases[0]]) s = storage.release_add([releases[0], releases[0]])
assert s == {} assert s == {}
s = storage.release_add([releases[0]]) s = storage.release_add([releases[0]])
assert s == {} assert s == {}
s = storage.release_add([releases[1]]) s = storage.release_add([releases[1]])
assert s == { assert s == {
"release:add": 1 + 1, "release:add": 1 + 1,
} }
missing_releases = storage.release_missing([releases[0]["id"], releases[1]["id"]]) missing_releases = storage.release_missing([r.id for r in releases])
assert list(missing_releases) == [] assert list(missing_releases) == []
s = storage.flush() s = storage.flush()
assert s == {} assert s == {}
def test_buffering_proxy_storage_clear(sample_data): def test_buffering_proxy_storage_clear(sample_data_model):
"""Clear operation on buffer """Clear operation on buffer
""" """
threshold = 10 threshold = 10
contents = sample_data["content"] contents = sample_data_model["content"]
assert 0 < len(contents) < threshold assert 0 < len(contents) < threshold
skipped_contents = sample_data["skipped_content"] skipped_contents = sample_data_model["skipped_content"]
assert 0 < len(skipped_contents) < threshold assert 0 < len(skipped_contents) < threshold
directories = sample_data["directory"] directories = sample_data_model["directory"]
assert 0 < len(directories) < threshold assert 0 < len(directories) < threshold
revisions = sample_data["revision"] revisions = sample_data_model["revision"]
assert 0 < len(revisions) < threshold assert 0 < len(revisions) < threshold
releases = sample_data["release"] releases = sample_data_model["release"]
assert 0 < len(releases) < threshold assert 0 < len(releases) < threshold
storage = get_storage_with_buffer_config( storage = get_storage_with_buffer_config(
min_batch_size={ min_batch_size={
"content": threshold, "content": threshold,
"skipped_content": threshold, "skipped_content": threshold,
"directory": threshold, "directory": threshold,
"revision": threshold, "revision": threshold,
▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines
About · Developer information · Legal