You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by ad...@apache.org on 2023/11/15 16:20:02 UTC
(jackrabbit-oak) branch issues/OAK-3349 updated (92d96801e2 -> 057ccb8555)
This is an automated email from the ASF dual-hosted git repository.
adulceanu pushed a change to branch issues/OAK-3349
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
discard 92d96801e2 OAK-3349 Partial Compaction
omit 5136d82b9a OAK-9949 Offline Tail Compaction
omit b375d8fa57 OAK-9922 Parallel Compaction Export package org.apache.jackrabbit.oak.plugins.index.counter from oak-core
omit 1d9095709b OAK-9922 Parallel Compaction
add 563c038a43 OAK-10452 - Mongo query that downloads ancestors of base path for regex filtering is doing a column scan (#1129)
add 8752a2eaa7 OAK-10429: Implementation for IncrementalFFS (#1123)
add 7be9852aa2 OAK-10456: log index name when bulk update fails (#1132)
add 189caed371 OAK-10450: bump testcontainers (#1131)
add ebd83ac64f OAK-10461: oak-search-elastic does not build under Java 17 - disable Felix SCR plugin (#1134)
add ad64ecbcfb Set LZ4 as the default compression algorithm for the indexing job. (#1133)
add cef242abb2 Revert "Set LZ4 as the default compression algorithm for the indexing job. (#1133)"
add 0b6e538736 OAK-10453 - Pipelined strategy: enforce size limit on memory taken by objects in the queue between download and transform thread (#1130)
add c6b4b08da6 OAK-9922: Parallel Compaction (#1043)
add 1f3cfa3b3d OAK-10465 : added io.netty netty-transport-native-unix-common embedded dependency for oak-segment-tar (#1137)
add f4343950c1 OAK-10458 - Set LZ4 as the default compression algorithm for the indexing job. (#1136)
add b2ffbb446c OAK-10463: Retrieve flatFileStore for indexing from baseFlatFileStore (#1135)
add c747535b92 OAK-9922: segment-tar: parallel compaction - remove oak-core dependency
add 8ef2cf89db OAK-10475 - Expose the Mongo client connection in MongoDocumentNodeStoreBuilderBase (#1142)
add 6b8aba2e76 OAK-10474: Release Oak 1.58.0 (#1145)
add 25568d29fe [maven-release-plugin] prepare release jackrabbit-oak-1.58.0
add 82ba25c3fc [maven-release-plugin] prepare for next development iteration
add 62a3f776cb OAK-10471: Implement ConflictHandler for UserPrincipalProvider Cache
add 763031951c OAK-10490 Suggest queries return duplicate entries if prefetch is enabled (#1148)
add 6a341e9dd1 OAK-10492: Update (shaded) Guava to 32.1.3 (#1153)
add ab772d73a9 OAK-10474: Release Oak 1.58.0 - version cleanup
add da19433ebe OAK-10486 : Resolution of inherited groups may terminate pre-maturely for external users (#1147)
add d7816c62b9 OAK-10491 - Indexing: pass a MongoDatabase instance instead of MongoConnection to indexing logic (#1151)
add 9c1fd98d68 OAK-10471 Implement ConflictHandler for UserPrincipalProvider Cache - fix tests (#1154)
add 13889a17d1 OAK-10495: Ignore DocumentStoreIndexerIT#parallelReindex* tests (#1157)
add 31a3341ac3 OAK-10487: replaced lsm estimator with cacheable count (#1149)
add 4c7e9f4144 OAK-10502: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 31.0 (#1162)
add 274398d483 OAK-10507: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 31.1 (#1163)
add 9aeef93e59 OAK-10508: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.0.0 (#1164)
add e3c2dd6303 OAK-10510: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.0.1 (#1167)
add 63aef2f4a1 OAK-10497 Properties order in FFS can be different across runs: option to sort (#1159)
add 1b31a22066 OAK-10512: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.1.0 (#1170)
add c255d16fed OAK-10497 Properties order in FFS can be different across runs: sort (#1175)
add 2a10b89f3e OAK-10513: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.1.1 (#1176)
add 9d54f6f23f OAK-10515: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.1.2 (#1177)
add e904e45942 OAK-10503 | Handle Exception in case of diff failures while building incremental FFS and handling Add operations during merging of incremental FFS. (#1172)
add bb98a83554 OAK-10516: oak-segment-azure/oak-blob-cloud-azure: update embedded Guava to 32.1.3 (#1178)
add 8d307fcdaa OAK-6760: Convert oak-blob-cloud to OSGi R6 annotations (#1144)
add d9ab316b2b OAK-10505 - Make PIPELINED the default indexing strategy. (#1168)
add abcfb39909 Deprecate all Mongo download strategies except for Pipelined. (#1169)
add b0ec30536f OAK-10514: Utility method to remove unmerged branch changes
add 9377d05a39 Merge pull request #1173 from mreutegg/OAK-10514
add 380467dbfe OAK-10504- Add log messages with start and end of the indexing job. (#1165)
add 9013717676 OAK-301 Improve documentation - fix the lucene documentation
add e682bc06c6 OAK-301 Improve documentation - fix the lucene documentation
add 546e889da6 OAK-10460 - PIPELINED strategy: support eager merging of intermediate sorted files (#1156)
add a4ca2a5b7a OAK-10518 IndexInfo should have a isActive() method (#1180)
add 7b650db0aa OAK-9705: fix explain output for elastic queries (#1161)
add 9014aff808 Issue/oak 9459b (#1183)
add 134b5af793 OAK-10460 - PIPELINED strategy: support eager merging of intermediate sorted files - fix test issue on Windows platform
add f12b415983 OAK-10478: Update Mockito dependency to 5.x
add ca7ea054c2 OAK-10524: SameNameSiblingTest: add (failing) test for getName() semantics
add facfe0e539 OAK-10521: Update Easymock dependency to 5.2
add ae576b4646 OAK-10464: Use Testcontainers instead of com.arakelian:docker-junit-rule (#1143)
add 84bf4f2d78 OAK-10517 : Consistently clean membership when switch between default and dynamic sync
add 893c0e4bb6 OAK-10525 : DefaultSyncContext.createValues : return value should be annotated with @NotNull
add 30a9407eec OAK-10462: o.a.j.o.plugins.version.VersionEditor#propertyAdded() may mistakenly assume an ongoing restore operation. (#1141)
add 68058e85bd OAK-10529: oak-run server: update Jetty to 9.4.53.v20231009 (#1188)
add 1d17b0e524 OAK-10522: o.a.j.o.index.ReindexIT#reindexIgnoreMissingTikaDepThrow() fails with Java 21
add a9307fc658 OAK-9949 Offline Tail Compaction (#1166)
add 9f8b7d8502 OAK-10526 : test to reproduce split doc GC with referenced revisions (#1184)
add 8d0d3c40c7 OAK-10420 Tool to compare Lucene index content (#1086)
add 0de5e850ec OAK-10511 - Get blobId without network roundtrip (#1171)
add 664ae24457 OAK-10537: Update Oak trunk and Oak 1.22 to Jackrabbit 2.20.13 (#1192)
add 04f3c97136 OAK-10527 Improve readability of the explain query output (#1187)
add 04e3b7d85a OAK-10543 skip writeAndReadManySegments test
add 873434545c OAK-10543 added issue number to the ignore declaration
add ef27c970ea OAK-10543 ignored writeAndReadManySegments in the abstract class
add e2a3d46934 OAK-10545: oak-jcr: add test for observing effect of remapping a namespace prefix to a different namespace name (#1196)
add d7edc02a09 OAK-10538 - Pipeline strategy: eliminate unnecessary intermediate copy of entries in transform stage (#1194)
add b138cf736c OAK-10545: oak-jcr: add test for observing effect of remapping a namespace prefix to a different namespace name - simplified not to use spi-commons
add 392ede19fe OAK-10547 - Fix: Indexing job fails at the end of reindexing if it took more than 24h to run (#1203)
add d1cd2bc18e OAK-10539: oak-search-elastic: migrate ingestion from Rest High Level Client to the new Java API Client (#1193)
add 5be575223d OAK-10526 : set split doc maxRev to 'now' at split time, to avoid it being GCed too early
add d177262672 Merge pull request #1190 from apache/OAK-10526
add d28e5b3c41 OAK-10539: fix oak-search-elastic bundle import packages (#1207)
add 0a0b252668 OAK-10550: add oak-search-elastic bundle test in oak-it-osgi (#1208)
add e73cb97510 OAK-10526 : fix test regression for mongo 70 split doc (#1209)
add 9d3d31127d OAK-10548: bump zookeeper to 3.9.1, re-organized optional import packages (#1205)
add 299002b63b OAK-6757: Convert oak-auth-ldap to OSGi R6 annotations (#1197)
add 0ba9cc4f9d OAK-10542 : test reproducing ignorance of split deleted properties (#1195)
new 057ccb8555 OAK-3349 Partial Compaction
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (92d96801e2)
\
N -- N -- N refs/heads/issues/OAK-3349 (057ccb8555)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
RELEASE-NOTES.txt | 100 ++-
oak-api/pom.xml | 2 +-
oak-auth-external/pom.xml | 2 +-
.../external/basic/DefaultSyncContext.java | 11 +-
.../external/basic/package-info.java | 2 +-
.../external/impl/DynamicSyncContext.java | 19 +-
.../external/impl/ExternalIdentityConstants.java | 9 +
.../principal/InheritedMembershipIterator.java | 29 +-
.../external/impl/DynamicSyncContextTest.java | 5 +-
.../external/impl/DynamicSyncTest.java | 66 ++
.../external/impl/SwitchSyncModeTest.java | 161 +++++
oak-auth-ldap/pom.xml | 7 +-
.../ldap/impl/LdapIdentityProvider.java | 21 +-
.../ldap/impl/LdapProviderConfig.java | 680 ++++++++-------------
oak-authorization-cug/pom.xml | 2 +-
oak-authorization-principalbased/pom.xml | 2 +-
oak-benchmarks-elastic/pom.xml | 2 +-
oak-benchmarks-lucene/pom.xml | 2 +-
oak-benchmarks-solr/pom.xml | 2 +-
oak-benchmarks/pom.xml | 2 +-
oak-blob-cloud-azure/pom.xml | 20 +-
.../cloud/azure/blobstorage/AzuriteDockerRule.java | 128 ++--
oak-blob-cloud/pom.xml | 6 +-
.../blob/cloud/s3/AbstractS3DataStoreService.java | 16 -
.../oak/blob/cloud/s3/S3DataStoreService.java | 19 +-
.../blob/cloud/s3/SharedS3DataStoreService.java | 19 +-
oak-blob-plugins/pom.xml | 2 +-
oak-blob/pom.xml | 2 +-
oak-commons/pom.xml | 2 +-
.../jackrabbit/oak/commons/sort/ExternalSort.java | 231 +++++--
.../oak/commons/sort/ExternalSortByteArray.java | 223 +++++++
.../jackrabbit/oak/commons/sort/package-info.java | 2 +-
.../commons/sort/ExternalSortByteArrayTest.java | 183 ++++++
.../oak/commons/sort/ExternalSortTest.java | 293 +++++----
.../jackrabbit/oak/commons/sort/test-file-2.csv | 6 +
oak-core-spi/pom.xml | 2 +-
oak-core/pom.xml | 3 +-
.../oak/plugins/index/FormattingUtils.java | 21 +-
.../jackrabbit/oak/plugins/index/IndexInfo.java | 11 +
.../oak/plugins/index/IndexInfoServiceImpl.java | 26 +-
.../jackrabbit/oak/plugins/index}/IndexName.java | 3 +-
.../oak/plugins/index/IndexPathService.java | 9 +
.../oak/plugins/index/IndexPathServiceImpl.java | 7 +-
.../oak/plugins/index/inventory/IndexPrinter.java | 1 +
.../oak/plugins/index/nodetype/NodeTypeIndex.java | 5 +-
.../index/property/PropertyIndexInfoProvider.java | 10 +
.../plugins/index/property/PropertyIndexPlan.java | 22 +-
.../plugins/index/reference/ReferenceIndex.java | 17 +-
.../oak/plugins/version/VersionEditor.java | 16 +-
.../oak/plugins/version/VersionableState.java | 84 ++-
.../jackrabbit/oak/query/QueryEngineImpl.java | 5 +-
.../jackrabbit/oak/query/QueryEngineSettings.java | 2 +-
.../jackrabbit/oak/query/QueryFormatter.java | 156 +++++
.../org/apache/jackrabbit/oak/query/QueryImpl.java | 4 -
.../jackrabbit/oak/query/ast/SelectorImpl.java | 9 +-
.../oak/query/index/TraversingIndex.java | 25 +-
.../oak/security/user/CacheConflictHandler.java | 95 +++
.../oak/security/user/UserConfigurationImpl.java | 2 +-
.../oak/plugins/index/FormattingUtilsTest.java | 88 +++
.../plugins/index/inventory/IndexPrinterTest.java | 10 +
.../index/nodetype/NodeTypeIndexQueryTest.java | 15 +-
.../index/property/MultiPropertyOrTest.java | 15 +-
.../index/property/OptionIndexTagTests.java | 8 +-
.../index/property/PropertyIndexDisabledTest.java | 13 +-
.../jackrabbit/oak/query/QueryFormatterTest.java | 183 ++++++
.../security/user/CacheConflictHandlerTest.java | 121 ++++
.../security/user/UserConfigurationImplTest.java | 2 +-
.../org/apache/jackrabbit/oak/query/sql2_index.txt | 253 +++++---
oak-doc-railroad-macro/pom.xml | 2 +-
oak-doc/pom.xml | 2 +-
oak-doc/src/site/markdown/query/lucene.md | 12 +-
oak-examples/pom.xml | 2 +-
oak-examples/standalone/pom.xml | 2 +-
oak-examples/webapp/pom.xml | 2 +-
oak-exercise/pom.xml | 2 +-
oak-http/pom.xml | 2 +-
oak-it-osgi/pom.xml | 8 +-
oak-it-osgi/test-bundles.xml | 1 +
oak-it/pom.xml | 6 +-
oak-jackrabbit-api/pom.xml | 2 +-
oak-jcr/pom.xml | 7 +-
.../apache/jackrabbit/oak/jcr/NamePathTest.java | 59 ++
.../jackrabbit/oak/jcr/SameNameSiblingTest.java | 13 +
.../oak/jcr/query/QueryFulltextTest.java | 7 +-
.../jackrabbit/oak/jcr/query/QueryPlanTest.java | 68 +--
.../apache/jackrabbit/oak/jcr/query/QueryTest.java | 25 +-
.../jackrabbit/oak/jcr/version/RestoreTest.java | 9 +
oak-lucene/pom.xml | 6 +-
.../index/lucene/LuceneIndexInfoProvider.java | 11 +
.../index/lucene/tools/IndexComparator.java | 247 ++++++++
.../CompositeNodeStoreLuceneIndexTest.java | 20 +-
.../oak/composite/CompositeNodeStoreQueryTest.java | 10 +-
.../composite/blueGreen/CustomizedIndexTest.java | 50 ++
.../oak/composite/blueGreen/Persistence.java | 6 +-
.../index/lucene/LuceneIndexAugmentTest.java | 12 +-
.../index/lucene/LuceneIndexQueryCommonTest.java | 3 +-
.../lucene/LucenePropertyIndexCommonTest.java | 4 +-
.../index/lucene/LucenePropertyIndexTest.java | 117 ++--
.../dynamicBoost/LuceneDynamicBoostTest.java | 5 +-
.../property/SynchronousPropertyIndexTest.java | 4 +-
oak-parent/pom.xml | 18 +-
oak-pojosr/pom.xml | 6 +-
oak-query-spi/pom.xml | 2 +-
oak-run-commons/pom.xml | 6 +-
.../jackrabbit/oak/index/IndexerSupport.java | 57 +-
.../indexer/document/DocumentStoreIndexerBase.java | 81 ++-
.../flatfile/FlatFileNodeStoreBuilder.java | 149 +++--
.../document/flatfile/FlatFileSplitter.java | 15 +-
.../indexer/document/flatfile/FlatFileStore.java | 41 +-
.../document/flatfile/FlatFileStoreUtils.java | 10 +-
.../MultithreadedTraverseWithSortStrategy.java | 57 +-
.../document/flatfile/NodeStateEntryReader.java | 6 +-
.../document/flatfile/NodeStateEntryWriter.java | 35 +-
.../document/flatfile/SimpleNodeStateHolder.java | 2 +-
.../indexer/document/flatfile/SortStrategy.java | 4 +
.../document/flatfile/StoreAndSortStrategy.java | 47 +-
.../flatfile/TraverseWithSortStrategy.java | 51 +-
.../flatfile/pipelined/NodeDocumentCodec.java | 161 +++++
.../pipelined/NodeDocumentCodecProvider.java | 45 ++
.../flatfile/pipelined/NodeStateEntryBatch.java | 47 +-
.../flatfile/pipelined/NodeStateHolder.java | 23 +-
.../flatfile/pipelined/NodeStateHolderFactory.java | 84 +++
.../flatfile/pipelined/PathElementComparator.java | 7 +-
.../flatfile/pipelined/PipelinedMergeSortTask.java | 273 +++++++--
.../pipelined/PipelinedMongoDownloadTask.java | 156 ++---
.../flatfile/pipelined/PipelinedSortBatchTask.java | 25 +-
.../flatfile/pipelined/PipelinedStrategy.java | 282 +++++----
.../flatfile/pipelined/PipelinedTransformTask.java | 118 ++--
.../incrementalstore/IncrementalFlatFileStore.java | 47 ++
.../IncrementalFlatFileStoreEditor.java | 117 ++++
...rementalFlatFileStoreNodeStateEntryReader.java} | 15 +-
...crementalFlatFileStoreNodeStateEntryWriter.java | 70 +++
.../IncrementalFlatFileStoreStrategy.java | 174 ++++++
.../IncrementalIndexStoreMetadata.java | 64 ++
.../IncrementalIndexStoreSortStrategy.java} | 22 +-
.../incrementalstore/IncrementalStoreBuilder.java | 147 +++++
.../IncrementalStoreOperand.java} | 25 +-
.../MergeIncrementalFlatFileStore.java | 223 +++++++
.../incrementalstore/MergeIncrementalStore.java | 13 +-
.../IndexStore.java} | 23 +-
.../document/indexstore/IndexStoreMetadata.java | 56 ++
.../IndexStoreMetadataOperator.java} | 13 +-
.../indexstore/IndexStoreMetadataOperatorImpl.java | 69 +++
.../IndexStoreSortStrategy.java} | 27 +-
.../indexstore/IndexStoreSortStrategyBase.java | 94 +++
.../document/indexstore/IndexStoreUtils.java | 148 +++++
.../oak/indexversion/IndexVersionOperation.java | 2 +-
.../oak/indexversion/PurgeOldIndexVersion.java | 2 +-
.../oak/indexversion/PurgeOldVersionUtils.java | 2 +-
.../oak/run/cli/DocumentFixtureProvider.java | 2 +
.../flatfile/FlatFileNodeStoreBuilderTest.java | 64 +-
.../document/flatfile/FlatFileStoreTest.java | 7 +-
.../document/flatfile/MergeIncrementalFFSTest.java | 130 ++++
.../pipelined/NodeStateEntryBatchTest.java | 34 +-
.../document/flatfile/pipelined/PipelinedIT.java | 135 ++--
.../PipelinedMergeSortTaskParameterizedTest.java | 117 ++++
.../pipelined/PipelinedMergeSortTaskTest.java | 137 +++--
.../pipelined/PipelinedMergeSortTaskTestBase.java | 129 ++++
.../pipelined/PipelinedMongoDownloadTaskTest.java | 47 +-
.../pipelined/PipelinedSortBatchTaskTest.java | 55 +-
...entalFlatFileStoreNodeStateEntryWriterTest.java | 71 +++
.../incrementalstore/IncrementalStoreIT.java | 497 +++++++++++++++
.../src/test/resources/logback-test.xml | 2 +
oak-run-elastic/pom.xml | 5 +-
.../indexversion/ElasticIndexVersionOperation.java | 2 +-
.../indexversion/ElasticPurgeOldIndexVersion.java | 2 +-
.../index/indexer/document/ElasticIndexerTest.java | 25 +-
.../ElasticPurgeOldIndexVersionTest.java | 3 +-
oak-run/pom.xml | 8 +-
.../apache/jackrabbit/oak/index/IndexCommand.java | 15 +-
.../oak/index/merge/IndexDefMergerUtils.java | 2 +-
.../jackrabbit/oak/index/merge/IndexDiff.java | 2 +-
.../indexversion/LuceneIndexVersionOperation.java | 2 +-
.../indexversion/LucenePurgeOldIndexVersion.java | 3 +-
oak-run/src/main/js/oak-mongo.js | 77 +++
.../oak/index/DocumentStoreIndexerIT.java | 142 ++---
.../org/apache/jackrabbit/oak/index/ReindexIT.java | 59 +-
.../apache/jackrabbit/oak/run/JsonIndexTest.java | 23 +-
oak-search-elastic/pom.xml | 107 +---
.../plugins/index/elastic/ElasticConnection.java | 23 +-
.../index/elastic/ElasticIndexDefinition.java | 12 +-
.../index/elastic/ElasticIndexInfoProvider.java | 10 +
.../index/elastic/ElasticIndexStatistics.java | 44 +-
.../elastic/index/ElasticBulkProcessorHandler.java | 241 ++++----
.../index/elastic/index/ElasticCustomAnalyzer.java | 8 +-
.../index/ElasticCustomAnalyzerMappings.java | 4 +-
.../index/elastic/index/ElasticDocument.java | 138 ++---
.../index/elastic/index/ElasticIndexHelper.java | 41 +-
.../index/elastic/index/ElasticIndexWriter.java | 47 +-
.../plugins/index/elastic/query/ElasticIndex.java | 41 +-
.../index/elastic/query/ElasticQueryIterator.java | 17 +-
.../index/elastic/query/ElasticRequestHandler.java | 2 +-
.../elastic/query/ElasticSpellcheckIterator.java | 71 ++-
.../elastic/query/ElasticSuggestIterator.java | 29 +-
.../query/async/ElasticResultRowAsyncIterator.java | 40 +-
.../index/elastic/ElasticDynamicBoostTest.java | 15 +-
.../index/elastic/ElasticIndexQueryCommonTest.java | 20 +-
.../index/elastic/ElasticIndexStatisticsTest.java | 16 +
.../elastic/ElasticIndexSuggestionCommonTest.java | 26 +
.../index/elastic/ElasticPropertyIndexTest.java | 12 +-
.../index/elastic/ElasticReliabilityTest.java | 24 +-
.../index/elastic/ElasticSpellcheckCommonTest.java | 28 +
.../plugins/index/elastic/ElasticTestServer.java | 7 +-
.../index/ElasticBulkProcessorHandlerTest.java | 17 +-
.../elastic/index/ElasticIndexWriterTest.java | 48 +-
oak-search-mt/pom.xml | 2 +-
oak-search/pom.xml | 2 +-
.../index/search/spi/query/FulltextIndex.java | 191 +++---
.../search/spi/query/FulltextIndexPlanner.java | 4 +-
.../plugins/index/search/spi/query/IndexName.java | 13 +-
.../oak/plugins/index/DynamicBoostCommonTest.java | 17 +-
.../oak/plugins/index/FunctionIndexCommonTest.java | 42 +-
.../oak/plugins/index/IndexQueryCommonTest.java | 6 +-
.../plugins/index/IndexSuggestionCommonTest.java | 22 +-
.../oak/plugins/index/OrderByCommonTest.java | 24 +-
.../oak/plugins/index/PropertyIndexCommonTest.java | 16 +-
.../oak/plugins/index/SpellcheckCommonTest.java | 4 +-
.../index/search/spi/query/FulltextIndexTest.java | 161 ++++-
.../index/search/spi/query/IndexNameTest.java | 1 +
oak-security-spi/pom.xml | 2 +-
.../AbstractAccessControlManagerTest.java | 10 +-
oak-segment-aws/pom.xml | 7 +-
oak-segment-azure/pom.xml | 19 +-
oak-segment-remote/pom.xml | 2 +-
.../AbstractPersistentCacheTest.java | 2 +
.../persistentcache/PersistentRedisCacheTest.java | 1 -
oak-segment-tar/pom.xml | 10 +-
.../jackrabbit/oak/segment/ApproximateCounter.java | 186 ++++++
.../oak/segment/DefaultSegmentWriter.java | 8 +
.../jackrabbit/oak/segment/ParallelCompactor.java | 1 -
.../oak/segment/DefaultSegmentWriterTest.java | 40 +-
.../oak/segment/test/TemporaryBlobStore.java | 33 +-
.../oak/segment/test/TemporaryFileStore.java | 37 +-
oak-shaded-guava/pom.xml | 6 +-
oak-solr-core/pom.xml | 2 +-
oak-solr-osgi/pom.xml | 50 +-
oak-store-composite/pom.xml | 2 +-
oak-store-document/pom.xml | 7 +-
.../oak/plugins/document/SplitOperations.java | 16 +-
.../mongo/MongoDocumentNodeStoreBuilderBase.java | 21 +-
.../plugins/document/MongoConnectionFactory.java | 2 +-
.../plugins/document/VersionGCSplitDocTest.java | 17 +-
.../document/VersionGarbageCollectorIT.java | 248 ++++++++
.../document/mongo/LeaseUpdateSocketTimeoutIT.java | 13 +-
.../plugins/document/mongo/MongoDockerRule.java | 124 ++--
oak-store-spi/pom.xml | 2 +-
oak-upgrade/pom.xml | 10 +-
pom.xml | 2 +-
248 files changed, 8738 insertions(+), 2940 deletions(-)
create mode 100644 oak-auth-external/src/test/java/org/apache/jackrabbit/oak/spi/security/authentication/external/impl/SwitchSyncModeTest.java
create mode 100644 oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortByteArray.java
create mode 100644 oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortByteArrayTest.java
copy {oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/spi/query => oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index}/IndexName.java (98%)
create mode 100644 oak-core/src/main/java/org/apache/jackrabbit/oak/query/QueryFormatter.java
create mode 100644 oak-core/src/main/java/org/apache/jackrabbit/oak/security/user/CacheConflictHandler.java
create mode 100644 oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/FormattingUtilsTest.java
create mode 100644 oak-core/src/test/java/org/apache/jackrabbit/oak/query/QueryFormatterTest.java
create mode 100644 oak-core/src/test/java/org/apache/jackrabbit/oak/security/user/CacheConflictHandlerTest.java
create mode 100644 oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/tools/IndexComparator.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/NodeDocumentCodec.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/NodeDocumentCodecProvider.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/NodeStateHolderFactory.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStore.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreEditor.java
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => incrementalstore/IncrementalFlatFileStoreNodeStateEntryReader.java} (66%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreNodeStateEntryWriter.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreStrategy.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalIndexStoreMetadata.java
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => incrementalstore/IncrementalIndexStoreSortStrategy.java} (65%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalStoreBuilder.java
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => incrementalstore/IncrementalStoreOperand.java} (69%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/MergeIncrementalFlatFileStore.java
copy oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexPathService.java => oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/MergeIncrementalStore.java (79%)
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => indexstore/IndexStore.java} (64%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreMetadata.java
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => indexstore/IndexStoreMetadataOperator.java} (66%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreMetadataOperatorImpl.java
copy oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/{flatfile/SortStrategy.java => indexstore/IndexStoreSortStrategy.java} (59%)
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreSortStrategyBase.java
create mode 100644 oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreUtils.java
create mode 100644 oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/MergeIncrementalFFSTest.java
create mode 100644 oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMergeSortTaskParameterizedTest.java
create mode 100644 oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMergeSortTaskTestBase.java
create mode 100644 oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreNodeStateEntryWriterTest.java
create mode 100644 oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalStoreIT.java
copy oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/package-info.java => oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/query/ElasticQueryIterator.java (58%)
create mode 100644 oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ApproximateCounter.java
(jackrabbit-oak) 01/01: OAK-3349 Partial Compaction
Posted by ad...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
adulceanu pushed a commit to branch issues/OAK-3349
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit 057ccb855519f85f00abc8f13fa7a7f09be0fcb6
Author: Lucas Weitzendorf <lw...@adobe.com>
AuthorDate: Fri Nov 11 15:17:24 2022 +0100
OAK-3349 Partial Compaction
---
.../site/markdown/nodestore/segment/overview.md | 22 +-
.../oak/backup/impl/FileStoreBackupImpl.java | 13 +-
.../oak/backup/impl/FileStoreRestoreImpl.java | 17 +-
.../oak/segment/CheckpointCompactor.java | 274 +++++++++----------
.../jackrabbit/oak/segment/ClassicCompactor.java | 191 +++++++------
.../apache/jackrabbit/oak/segment/Compactor.java | 67 ++++-
.../jackrabbit/oak/segment/CompactorUtils.java | 3 +-
.../jackrabbit/oak/segment/ParallelCompactor.java | 303 ++++++++++++---------
.../segment/{file => }/SegmentWriterFactory.java | 7 +-
.../segment/file/AbstractCompactionStrategy.java | 198 +++++++++-----
.../file/AbstractGarbageCollectionStrategy.java | 18 +-
.../file/CleanupFirstCompactionStrategy.java | 118 ++------
.../oak/segment/file/CompactedNodeState.java | 69 +++++
.../oak/segment/file/CompactionResult.java | 44 ++-
.../oak/segment/file/CompactionStrategy.java | 8 +-
.../oak/segment/file/CompactionWriter.java | 105 +++++++
.../oak/segment/file/DefaultCleanupContext.java | 106 +++++++
.../oak/segment/file/DefaultCleanupStrategy.java | 63 +----
.../oak/segment/file/FullCompactionStrategy.java | 7 +-
.../jackrabbit/oak/segment/file/GCIncrement.java | 66 +++++
.../oak/segment/file/GCNodeWriteMonitor.java | 7 +-
.../segment/file/GarbageCollectionStrategy.java | 1 +
.../oak/segment/file/GarbageCollector.java | 2 +
.../jackrabbit/oak/segment/file/Reclaimers.java | 15 +-
.../oak/segment/file/TailCompactionStrategy.java | 7 +-
.../oak/segment/file/cancel/Canceller.java | 13 +-
.../segment/file/cancel/ShortCircuitCanceller.java | 6 +-
.../oak/segment/file/tar/GCGeneration.java | 20 +-
.../jackrabbit/oak/segment/file/tar/TarReader.java | 3 +-
.../segment/AbstractCompactorExternalBlobTest.java | 32 ++-
.../oak/segment/AbstractCompactorTest.java | 106 ++++++-
.../CheckpointCompactorExternalBlobTest.java | 22 +-
.../oak/segment/CheckpointCompactorTest.java | 28 +-
.../oak/segment/ClassicCompactorTest.java | 16 +-
.../jackrabbit/oak/segment/CompactorTestUtils.java | 14 +-
.../segment/ParallelCompactorExternalBlobTest.java | 29 +-
.../oak/segment/ParallelCompactorTest.java | 39 ++-
.../oak/segment/file/CompactionStrategyTest.java | 56 ----
.../file/DefaultGarbageCollectionStrategyTest.java | 113 ++++++++
.../segment/file/FullCompactionStrategyTest.java | 168 ++++++++++++
.../oak/segment/file/GCIncrementTest.java | 63 +++++
41 files changed, 1684 insertions(+), 775 deletions(-)
diff --git a/oak-doc/src/site/markdown/nodestore/segment/overview.md b/oak-doc/src/site/markdown/nodestore/segment/overview.md
index 499bcb3cf2..cd2133c058 100644
--- a/oak-doc/src/site/markdown/nodestore/segment/overview.md
+++ b/oak-doc/src/site/markdown/nodestore/segment/overview.md
@@ -93,12 +93,12 @@ If there is not enough garbage to justify the creation of a new generation, the
If the output of this phase reports that the amount of garbage is beyond a certain threshold, the system creates a new generation and goes on with the next phase.
Compaction executes after a new generation is created.
-The purpose of compaction is to create a compact representation of the current generation. For this the current generation is copied to the new generation leaving out anything from the current generation that is not reachable anymore. Starting with Oak 1.8 compaction can operate in either of two modes: full compaction and tail compaction. Full compaction copies all revisions pertaining to the current generation to the new generation. In contrast tail compaction only copies the most recen [...]
+The purpose of compaction is to create a compact representation of the current generation. For this the current generation is copied to the new generation leaving out anything from the current generation that is not reachable anymore. Starting with Oak 1.8 compaction can operate in either of two modes: full compaction and tail compaction. Full compaction copies all revisions pertaining to the current generation to the new generation. In contrast tail compaction only copies the most recen [...]
Cleanup is the last phase of garbage collection and kicks in as soon as compaction is done.
Once relevant data is safe in the new generation, old and unused data from a previous generation can be removed.
This phase locates outdated pieces of data from one of the oldest generations and removes it from the system.
-This is the only phase where data is actually deleted and disk space is finally freed. The amount of freed disk space depends on the preceding compaction operation. In general cleanup can free less space after a tail compaction than after a full compaction. However, this only becomes effective a further garbage collection cycle due to the system always retaining a total of two generations.
+This is the only phase where data is actually deleted and disk space is finally freed. The amount of freed disk space depends on the preceding compaction operation. In general cleanup can free less space after a tail compaction than after a full compaction. However, this usually only becomes effective after a further garbage collection cycle as the system retains a total of two generations by default.
### <a name="offline-garbage-collection"/> Offline Garbage Collection
@@ -133,7 +133,7 @@ This section groups those log messages by function, so to provide a useful refer
Please note that the following messages are to be used as an example only.
To make the examples clear, some information like the date and time, the name of the thread, and the name of the logger are removed.
-These information depend on the configuration of your logging framework.
+This information depends on the configuration of your logging framework.
Moreover, some of those messages contain data that can and will change from one execution to the other.
Every log message generated during the garbage collection process includes a sequence number
@@ -291,17 +291,17 @@ TarMK GC #2: compacting root.
##### <a name="how-does-compaction-make-use-of-multithreading"/> How does compaction make use of multithreading?
The parallel compactor adds an initial exploration phase to the compaction process, which scans and splits the content tree
-into multiple parts to be processed simultaneously. For this to be efficient, the tree is only expanded until a certain
-number of nodes is reached, which is defined relative to the number of threads (main thread + compaction workers).
-
+into multiple parts to be processed simultaneously. For this to be efficient, the tree is only expanded until a pre-defined
+(currently 10,000) number of nodes is reached.
```
TarMK GC #2: compacting with 8 threads.
TarMK GC #2: exploring content tree to find subtrees for parallel compaction.
-TarMK GC #2: target node count for expansion is 7000, based on 7 available workers.
-TarMK GC #2: Found 3 nodes at depth 1, target is 7000.
-TarMK GC #2: Found 48 nodes at depth 2, target is 7000.
-TarMK GC #2: Found 663 nodes at depth 3, target is 7000.
-TarMK GC #2: Found 66944 nodes at depth 4, target is 7000.
+TarMK GC #2: target node count for expansion is 10000.
+TarMK GC #2: found 1 nodes at depth 0.
+TarMK GC #2: found 3 nodes at depth 1.
+TarMK GC #2: found 48 nodes at depth 2.
+TarMK GC #2: found 663 nodes at depth 3.
+TarMK GC #2: found 66944 nodes at depth 4.
```
##### <a name="how-does-compaction-works-with-concurrent-writes"/> How does compaction work with concurrent writes?
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreBackupImpl.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreBackupImpl.java
index d2a1adca63..d3144478f8 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreBackupImpl.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreBackupImpl.java
@@ -38,6 +38,7 @@ import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.FileStoreBuilder;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
@@ -86,15 +87,11 @@ public class FileStoreBackupImpl implements FileStoreBackup {
bufferWriter,
backup.getBinariesInlineThreshold()
);
- ClassicCompactor compactor = new ClassicCompactor(
- backup.getReader(),
- writer,
- backup.getBlobStore(),
- GCNodeWriteMonitor.EMPTY
- );
+ CompactionWriter compactionWriter = new CompactionWriter(backup.getReader(), backup.getBlobStore(), gen, writer);
+ ClassicCompactor compactor = new ClassicCompactor(compactionWriter, GCNodeWriteMonitor.EMPTY);
SegmentNodeState head = backup.getHead();
- SegmentNodeState after = compactor.compact(head, current, head, Canceller.newCanceller());
- writer.flush();
+ SegmentNodeState after = compactor.compactUp(head, current, Canceller.newCanceller());
+ compactionWriter.flush();
if (after != null) {
backup.getRevisions().setHead(head.getRecordId(), after.getRecordId());
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreRestoreImpl.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreRestoreImpl.java
index c4b04ba67a..cbc0c84d28 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreRestoreImpl.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/backup/impl/FileStoreRestoreImpl.java
@@ -19,14 +19,12 @@
package org.apache.jackrabbit.oak.backup.impl;
-import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions.defaultGCOptions;
import static org.apache.jackrabbit.oak.segment.file.FileStoreBuilder.fileStoreBuilder;
import java.io.File;
import java.io.IOException;
import org.apache.jackrabbit.guava.common.base.Stopwatch;
-import org.apache.jackrabbit.guava.common.base.Suppliers;
import org.apache.jackrabbit.oak.backup.FileStoreRestore;
import org.apache.jackrabbit.oak.segment.DefaultSegmentWriter;
import org.apache.jackrabbit.oak.segment.ClassicCompactor;
@@ -34,9 +32,9 @@ import org.apache.jackrabbit.oak.segment.SegmentBufferWriter;
import org.apache.jackrabbit.oak.segment.SegmentNodeState;
import org.apache.jackrabbit.oak.segment.SegmentWriter;
import org.apache.jackrabbit.oak.segment.WriterCacheManager;
-import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
import org.apache.jackrabbit.oak.segment.file.ReadOnlyFileStore;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
@@ -82,15 +80,10 @@ public class FileStoreRestoreImpl implements FileStoreRestore {
bufferWriter,
store.getBinariesInlineThreshold()
);
- SegmentGCOptions gcOptions = defaultGCOptions().setOffline();
- ClassicCompactor compactor = new ClassicCompactor(
- store.getReader(),
- writer,
- store.getBlobStore(),
- GCNodeWriteMonitor.EMPTY
- );
- SegmentNodeState after = compactor.compact(current, head, current, Canceller.newCanceller());
- writer.flush();
+ CompactionWriter compactionWriter = new CompactionWriter(store.getReader(), store.getBlobStore(), gen, writer);
+ ClassicCompactor compactor = new ClassicCompactor(compactionWriter, GCNodeWriteMonitor.EMPTY);
+ SegmentNodeState after = compactor.compactUp(current, head, Canceller.newCanceller());
+ compactionWriter.flush();
store.getRevisions().setHead(current.getRecordId(), after.getRecordId());
} finally {
restore.close();
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
index e9d63cec05..2b24d14de4 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
@@ -19,6 +19,7 @@
package org.apache.jackrabbit.oak.segment;
import static java.util.Objects.requireNonNull;
+import static org.apache.jackrabbit.guava.common.base.Preconditions.checkState;
import static org.apache.jackrabbit.guava.common.collect.Lists.newArrayList;
import static org.apache.jackrabbit.guava.common.collect.Maps.newHashMap;
import static org.apache.jackrabbit.guava.common.collect.Maps.newLinkedHashMap;
@@ -26,10 +27,12 @@ import static org.apache.jackrabbit.oak.commons.PathUtils.elements;
import static org.apache.jackrabbit.oak.commons.PathUtils.getName;
import static org.apache.jackrabbit.oak.commons.PathUtils.getParentPath;
import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
-import static org.apache.jackrabbit.oak.segment.CompactorUtils.getStableIdBytes;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -37,9 +40,10 @@ import java.util.Map.Entry;
import org.apache.jackrabbit.oak.commons.Buffer;
import org.apache.jackrabbit.oak.plugins.memory.MemoryChildNodeEntry;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
-import org.apache.jackrabbit.oak.spi.blob.BlobStore;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
import org.apache.jackrabbit.oak.spi.state.DefaultNodeStateDiff;
@@ -59,124 +63,132 @@ import org.jetbrains.annotations.Nullable;
* the same checkpoint or root state occur again in a later compaction retry cycle.</li>
* </ul>
*/
-public class CheckpointCompactor implements Compactor {
- @NotNull
- protected final GCMonitor gcListener;
+public class CheckpointCompactor extends Compactor {
+ protected final @NotNull GCMonitor gcListener;
- @NotNull
- private final Map<NodeState, NodeState> cpCache = newHashMap();
+ private final @NotNull Map<NodeState, CompactedNodeState> cpCache = new HashMap<>();
- @NotNull
- protected final ClassicCompactor compactor;
-
- @NotNull
- private final NodeWriter nodeWriter;
-
- private interface NodeWriter {
- @NotNull
- SegmentNodeState writeNode(@NotNull NodeState node, @Nullable Buffer stableId) throws IOException;
- }
+ protected final @NotNull ClassicCompactor compactor;
/**
* Create a new instance based on the passed arguments.
- * @param gcListener listener receiving notifications about the garbage collection process
- * @param reader segment reader used to read from the segments
- * @param writer segment writer used to serialise to segments
- * @param blobStore the blob store or {@code null} if none
- * @param compactionMonitor notification call back for each compacted nodes,
- * properties, and binaries
+ *
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @param writer segment writer used to serialise to segments
+ * @param compactionMonitor notification call back for each compacted nodes,
+ * properties, and binaries
*/
public CheckpointCompactor(
@NotNull GCMonitor gcListener,
- @NotNull SegmentReader reader,
- @NotNull SegmentWriter writer,
- @Nullable BlobStore blobStore,
+ @NotNull CompactionWriter writer,
@NotNull GCNodeWriteMonitor compactionMonitor) {
this.gcListener = gcListener;
- this.compactor = new ClassicCompactor(reader, writer, blobStore, compactionMonitor);
- this.nodeWriter = (node, stableId) -> {
- RecordId nodeId = writer.writeNode(node, stableId);
- return new SegmentNodeState(reader, writer, blobStore, nodeId);
- };
+ this.compactor = new ClassicCompactor(writer, compactionMonitor);
}
- /**
- * Compact {@code uncompacted} on top of an optional {@code base}.
- * @param base the base state to compact against
- * @param uncompacted the uncompacted state to compact
- * @param onto the state onto which to compact the change between {@code base} and
- * {@code uncompacted}
- * @return compacted clone of {@code uncompacted} or {@code null} if cancelled.
- * @throws IOException
- */
@Override
- @Nullable
- public SegmentNodeState compact(
- @NotNull NodeState base,
- @NotNull NodeState uncompacted,
- @NotNull NodeState onto,
- Canceller canceller
+ public @Nullable CompactedNodeState compactDown(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
) throws IOException {
- // Collect a chronologically ordered list of roots for the uncompacted
- // state. This list consists of all checkpoints followed by the root.
- LinkedHashMap<String, NodeState> uncompactedRoots = collectRoots(base, uncompacted);
-
- // Compact the list of uncompacted roots to a list of compacted roots.
- LinkedHashMap<String, NodeState> compactedRoots = compact(
- getRoot(base),
- uncompactedRoots,
- getRoot(onto),
- canceller
- );
+ Iterator<Entry<String, NodeState>> iterator = collectRoots(before, after).entrySet().iterator();
+ Entry<String, NodeState> entry = iterator.next();
+ String path = entry.getKey();
- if (compactedRoots == null) {
- return null;
+ // could already be in cache if compactor is reused
+ CompactedNodeState compacted = cpCache.get(entry.getValue());
+ gcListener.info("compacting {}.", path);
+ if (compacted == null) {
+ compacted = compactDownWithDelegate(getRoot(before), entry.getValue(), hardCanceller, softCanceller);
+ if (compacted == null) {
+ return null;
+ }
}
- // Build a compacted super root by replacing the uncompacted roots with
- // the compacted ones in the original node.
- NodeBuilder builder = uncompacted.builder();
- for (Entry<String, NodeState> compactedRoot : compactedRoots.entrySet()) {
- String path = compactedRoot.getKey();
- NodeState state = compactedRoot.getValue();
- NodeBuilder childBuilder = getChild(builder, getParentPath(path));
- childBuilder.setChildNode(getName(path), state);
+ NodeBuilder builder = after.builder();
+ Buffer stableIdBytes = requireNonNull(CompactorUtils.getStableIdBytes(after));
+
+ getChild(builder, getParentPath(path)).setChildNode(getName(path), compacted);
+
+ if (compacted.isComplete()) {
+ cpCache.put(entry.getValue(), compacted);
+ } else {
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, false);
}
- return nodeWriter.writeNode(builder.getNodeState(), requireNonNull(getStableIdBytes(uncompacted)));
- }
+ before = entry.getValue();
- @NotNull
- private static NodeState getRoot(@NotNull NodeState node) {
- return node.hasChildNode("root")
- ? node.getChildNode("root")
- : EMPTY_NODE;
+ while (iterator.hasNext()) {
+ entry = iterator.next();
+ path = entry.getKey();
+ gcListener.info("compacting {}.", path);
+
+ compacted = compactWithCache(before, entry.getValue(), compacted, hardCanceller);
+ if (compacted == null) {
+ return null;
+ }
+
+ before = entry.getValue();
+ checkState(compacted.isComplete());
+ getChild(builder, getParentPath(path)).setChildNode(getName(path), compacted);
+
+ if (softCanceller.check().isCancelled()) {
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, false);
+ }
+ }
+
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, true);
}
- /**
- * Compact a list of uncompacted roots on top of base roots of the same key or
- * an empty node if none.
- */
- @Nullable
- private LinkedHashMap<String, NodeState> compact(
- @NotNull NodeState base,
- @NotNull LinkedHashMap<String, NodeState> uncompactedRoots,
- @NotNull NodeState onto,
- Canceller canceller
+ @Override
+ public @Nullable CompactedNodeState compact(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull NodeState onto,
+ @NotNull Canceller canceller
) throws IOException {
- LinkedHashMap<String, NodeState> compactedRoots = newLinkedHashMap();
- for (Entry<String, NodeState> uncompactedRoot : uncompactedRoots.entrySet()) {
- String path = uncompactedRoot.getKey();
- NodeState uncompacted = uncompactedRoot.getValue();
- Result result = compactWithCache(base, uncompacted, onto, path, canceller);
- if (result == null) {
+ LinkedHashMap<String, NodeState> roots = collectRoots(before, after);
+
+ NodeBuilder builder = after.builder();
+ Buffer stableIdBytes = requireNonNull(CompactorUtils.getStableIdBytes(after));
+
+ before = getRoot(before);
+ onto = getRoot(onto);
+
+ for (Entry<String, NodeState> entry : roots.entrySet()) {
+ String path = entry.getKey();
+ after = entry.getValue();
+ CompactedNodeState compacted = compactWithCache(before, after, onto, canceller);
+ if (compacted == null) {
return null;
}
- base = result.nextBefore;
- onto = result.nextOnto;
- compactedRoots.put(path, result.compacted);
+ checkState(compacted.isComplete());
+ getChild(builder, getParentPath(path)).setChildNode(getName(path), compacted);
+ before = after;
+ onto = compacted;
}
- return compactedRoots;
+
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, true);
+ }
+
+ private @Nullable CompactedNodeState compactWithCache(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull NodeState onto,
+ @NotNull Canceller canceller
+ ) throws IOException {
+ CompactedNodeState compacted = cpCache.get(after);
+ if (compacted == null) {
+ compacted = compactWithDelegate(before, after, onto, canceller);
+ if (compacted != null) {
+ cpCache.put(after, compacted);
+ }
+ } else {
+ gcListener.info("found checkpoint in cache.");
+ }
+ return compacted;
}
/**
@@ -184,11 +196,10 @@ public class CheckpointCompactor implements Compactor {
* state from a {@code superRoot}. This list consists of all checkpoints followed by
* the root.
*/
- @NotNull
- private LinkedHashMap<String, NodeState> collectRoots(@NotNull NodeState superRootBefore, @NotNull NodeState superRootAfter) {
- LinkedHashMap<String, NodeState> roots = newLinkedHashMap();
-
- List<ChildNodeEntry> checkpoints = newArrayList();
+ private @NotNull LinkedHashMap<String, NodeState> collectRoots(
+ @NotNull NodeState superRootBefore,
+ @NotNull NodeState superRootAfter) {
+ List<ChildNodeEntry> checkpoints = new ArrayList<>();
superRootAfter.getChildNode("checkpoints").compareAgainstBaseState(
superRootBefore.getChildNode("checkpoints"), new DefaultNodeStateDiff() {
@Override
@@ -205,11 +216,12 @@ public class CheckpointCompactor implements Compactor {
return Long.compare(c1, c2);
});
+ LinkedHashMap<String, NodeState> roots = new LinkedHashMap<>();
for (ChildNodeEntry checkpoint : checkpoints) {
String name = checkpoint.getName();
NodeState node = checkpoint.getNodeState();
- gcListener.info("found checkpoint {} created at {}.",
- name, new Date(node.getLong("created")));
+ gcListener.info("found checkpoint {} created on {}.",
+ name, new Date(node.getLong("created")));
roots.put("checkpoints/" + name + "/root", node.getChildNode("root"));
}
roots.put("root", superRootAfter.getChildNode("root"));
@@ -217,65 +229,35 @@ public class CheckpointCompactor implements Compactor {
return roots;
}
- @NotNull
- private static NodeBuilder getChild(NodeBuilder builder, String path) {
+ private static @NotNull NodeState getRoot(@NotNull NodeState node) {
+ return node.hasChildNode("root") ? node.getChildNode("root") : EMPTY_NODE;
+ }
+
+ private static @NotNull NodeBuilder getChild(NodeBuilder builder, String path) {
for (String name : elements(path)) {
builder = builder.getChildNode(name);
}
return builder;
}
- private static class Result {
- final NodeState compacted;
- final NodeState nextBefore;
- final NodeState nextOnto;
-
- Result(@NotNull NodeState compacted, @NotNull NodeState nextBefore, @NotNull NodeState nextOnto) {
- this.compacted = compacted;
- this.nextBefore = nextBefore;
- this.nextOnto = nextOnto;
- }
- }
-
/**
* Delegate compaction to another, usually simpler, implementation.
*/
- @Nullable
- protected SegmentNodeState compactWithDelegate(
+ protected @Nullable CompactedNodeState compactDownWithDelegate(
@NotNull NodeState before,
@NotNull NodeState after,
- @NotNull NodeState onto,
- Canceller canceller
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
) throws IOException {
- return compactor.compact(before, after, onto, canceller);
+ return compactor.compactDown(before, after, hardCanceller, softCanceller);
}
- /**
- * Compact {@code after} against {@code before} on top of {@code onto} unless
- * {@code after} has been compacted before and is found in the cache. In this
- * case the cached version of the previously compacted {@code before} is returned.
- */
- @Nullable
- private Result compactWithCache(
- @NotNull NodeState before,
- @NotNull NodeState after,
- @NotNull NodeState onto,
- @NotNull String path,
- Canceller canceller
+ protected @Nullable CompactedNodeState compactWithDelegate(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull NodeState onto,
+ @NotNull Canceller canceller
) throws IOException {
- gcListener.info("compacting {}.", path);
- NodeState compacted = cpCache.get(after);
- if (compacted == null) {
- compacted = compactWithDelegate(before, after, onto, canceller);
- if (compacted == null) {
- return null;
- } else {
- cpCache.put(after, compacted);
- return new Result(compacted, after, compacted);
- }
- } else {
- gcListener.info("found {} in cache.", path);
- return new Result(compacted, before, onto);
- }
+ return compactor.compact(before, after, onto, canceller);
}
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ClassicCompactor.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ClassicCompactor.java
index 61a2fb67c3..2d5a3ea53a 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ClassicCompactor.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ClassicCompactor.java
@@ -28,6 +28,7 @@ import static org.apache.jackrabbit.oak.plugins.memory.MultiBinaryPropertyState.
import static org.apache.jackrabbit.oak.plugins.memory.PropertyStates.createProperty;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import org.apache.jackrabbit.oak.api.Blob;
@@ -35,9 +36,12 @@ import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Type;
import org.apache.jackrabbit.oak.commons.Buffer;
import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.plugins.memory.ModifiedNodeState;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
-import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
import org.jetbrains.annotations.NotNull;
@@ -51,7 +55,7 @@ import org.jetbrains.annotations.Nullable;
* A node can either be compacted on its own or alternatively the difference between
* two nodes can be compacted on top of an already compacted node.
*/
-public class ClassicCompactor implements Compactor {
+public class ClassicCompactor extends Compactor {
/**
* Number of content updates that need to happen before the updates
@@ -60,127 +64,130 @@ public class ClassicCompactor implements Compactor {
static final int UPDATE_LIMIT =
Integer.getInteger("compaction.update.limit", 10000);
- @NotNull
- private final SegmentWriter writer;
+ private final @NotNull CompactionWriter writer;
- @NotNull
- private final SegmentReader reader;
-
- @Nullable
- private final BlobStore blobStore;
-
- @NotNull
- private final GCNodeWriteMonitor compactionMonitor;
+ private final @NotNull GCNodeWriteMonitor compactionMonitor;
/**
* Create a new instance based on the passed arguments.
- * @param reader segment reader used to read from the segments
* @param writer segment writer used to serialise to segments
- * @param blobStore the blob store or {@code null} if none
* @param compactionMonitor notification call back for each compacted nodes,
* properties, and binaries
*/
public ClassicCompactor(
- @NotNull SegmentReader reader,
- @NotNull SegmentWriter writer,
- @Nullable BlobStore blobStore,
+ @NotNull CompactionWriter writer,
@NotNull GCNodeWriteMonitor compactionMonitor) {
this.writer = checkNotNull(writer);
- this.reader = checkNotNull(reader);
- this.blobStore = blobStore;
this.compactionMonitor = checkNotNull(compactionMonitor);
}
- /**
- * Compact a given {@code state}
- * @param state the node state to compact
- * @return the compacted node state or {@code null} if cancelled.
- * @throws IOException
- */
- @Nullable
- public SegmentNodeState compact(@NotNull NodeState state, Canceller canceller) throws IOException {
- return compact(EMPTY_NODE, state, EMPTY_NODE, canceller);
+ @Override
+ public @Nullable CompactedNodeState compactDown(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
+ ) throws IOException {
+ return compact(before, after, after, hardCanceller, softCanceller);
}
- /**
- * compact the differences between {@code after} and {@code before} on top of {@code ont}.
- * @param before the node state to diff against from {@code after}
- * @param after the node state diffed against {@code before}
- * @param onto the node state compacted onto
- * @return the compacted node state or {@code null} if cancelled.
- * @throws IOException
- */
- @Nullable
- public SegmentNodeState compact(
+ @Override
+ public @Nullable CompactedNodeState compact(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull NodeState onto,
+ @NotNull Canceller canceller
+ ) throws IOException {
+ return compact(before, after, onto, canceller, Canceller.newCanceller());
+ }
+
+ private @Nullable CompactedNodeState compact(
@NotNull NodeState before,
@NotNull NodeState after,
@NotNull NodeState onto,
- Canceller canceller
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
) throws IOException {
- checkNotNull(before);
- checkNotNull(after);
- checkNotNull(onto);
- return new CompactDiff(onto, canceller).diff(before, after);
+ CompactedNodeState compactedState = getPreviouslyCompactedState(after);
+ if (compactedState == null) {
+ compactedState = new CompactDiff(onto, hardCanceller, softCanceller).diff(before, after);
+ }
+ return compactedState;
}
- protected SegmentNodeState writeNodeState(NodeState nodeState, Buffer stableIdBytes) throws IOException {
- RecordId nodeId = writer.writeNode(nodeState, stableIdBytes);
- compactionMonitor.onNode();
- return new SegmentNodeState(reader, writer, blobStore, nodeId);
+ protected @Nullable CompactedNodeState writeNodeState(
+ @NotNull NodeState nodeState,
+ @Nullable Buffer stableIdBytes,
+ boolean complete
+ ) throws IOException {
+ if (complete) {
+ CompactedNodeState compacted = writer.writeFullyCompactedNode(nodeState, stableIdBytes);
+ compactionMonitor.onNode();
+ return compacted;
+ } else {
+ return writer.writePartiallyCompactedNode(nodeState, stableIdBytes);
+ }
}
- private class CompactDiff implements NodeStateDiff {
- @NotNull
- private MemoryNodeBuilder builder;
-
- @NotNull
- private final NodeState base;
-
- private final Canceller canceller;
-
- @Nullable
- private IOException exception;
+ protected @Nullable CompactedNodeState getPreviouslyCompactedState(NodeState nodeState) {
+ return writer.getPreviouslyCompactedState(nodeState);
+ }
+ private class CompactDiff implements NodeStateDiff {
+ private final @NotNull NodeState base;
+ private final @NotNull Canceller hardCanceller;
+ private final @NotNull Canceller softCanceller;
+ private final @NotNull List<PropertyState> modifiedProperties = new ArrayList<>();
+ private @NotNull NodeBuilder builder;
+ private @Nullable IOException exception;
private long modCount;
private void updated() throws IOException {
if (++modCount % UPDATE_LIMIT == 0) {
- RecordId newBaseId = writer.writeNode(builder.getNodeState(), null);
- SegmentNodeState newBase = new SegmentNodeState(reader, writer, blobStore, newBaseId);
+ SegmentNodeState newBase = writeNodeState(builder.getNodeState(), null, false);
+ checkNotNull(newBase);
builder = new MemoryNodeBuilder(newBase);
}
}
- CompactDiff(@NotNull NodeState base, Canceller canceller) {
- this.builder = new MemoryNodeBuilder(checkNotNull(base));
- this.canceller = canceller;
- this.base = base;
+ CompactDiff(@NotNull NodeState base, @NotNull Canceller hardCanceller, @NotNull Canceller softCanceller) {
+ this.base = checkNotNull(base);
+ this.builder = new MemoryNodeBuilder(base);
+ this.hardCanceller = checkNotNull(hardCanceller);
+ this.softCanceller = checkNotNull(softCanceller);
+ }
+
+ private @NotNull CancelableDiff newCancelableDiff() {
+ return new CancelableDiff(this, () ->
+ softCanceller.check().isCancelled() || hardCanceller.check().isCancelled());
}
- @Nullable
- SegmentNodeState diff(@NotNull NodeState before, @NotNull NodeState after) throws IOException {
- boolean success = after.compareAgainstBaseState(before,
- new CancelableDiff(this, () -> canceller.check().isCancelled()));
+ @Nullable CompactedNodeState diff(@NotNull NodeState before, @NotNull NodeState after) throws IOException {
+ boolean success = after.compareAgainstBaseState(before, newCancelableDiff());
if (exception != null) {
throw new IOException(exception);
} else if (success) {
+ // delay property compaction until the end in case compaction is cancelled
+ modifiedProperties.forEach(property -> builder.setProperty(compact(property)));
NodeState nodeState = builder.getNodeState();
- checkState(modCount == 0 || !(nodeState instanceof SegmentNodeState));
- return writeNodeState(nodeState, CompactorUtils.getStableIdBytes(after));
- } else {
+ checkState(modCount == 0 || nodeState instanceof ModifiedNodeState);
+ return writeNodeState(nodeState, CompactorUtils.getStableIdBytes(after), true);
+ } else if (hardCanceller.check().isCancelled()) {
return null;
+ } else {
+ return writeNodeState(builder.getNodeState(), CompactorUtils.getStableIdBytes(after), false);
}
}
@Override
public boolean propertyAdded(@NotNull PropertyState after) {
- builder.setProperty(compact(after));
+ modifiedProperties.add(after);
return true;
}
@Override
public boolean propertyChanged(@NotNull PropertyState before, @NotNull PropertyState after) {
- builder.setProperty(compact(after));
+ modifiedProperties.add(after);
return true;
}
@@ -190,38 +197,31 @@ public class ClassicCompactor implements Compactor {
return true;
}
- @Override
- public boolean childNodeAdded(@NotNull String name, @NotNull NodeState after) {
+ private boolean childNodeUpdated(@NotNull String name, @NotNull NodeState before, @NotNull NodeState after) {
try {
- SegmentNodeState compacted = compact(after, canceller);
- if (compacted != null) {
- updated();
- builder.setChildNode(name, compacted);
- return true;
- } else {
+ NodeState child = base.getChildNode(name);
+ NodeState onto = child.exists() ? child : EMPTY_NODE;
+ CompactedNodeState compacted = compact(before, after, onto, hardCanceller, softCanceller);
+ if (compacted == null) {
return false;
}
+ updated();
+ builder.setChildNode(name, compacted);
+ return compacted.isComplete();
} catch (IOException e) {
exception = e;
return false;
}
}
+ @Override
+ public boolean childNodeAdded(@NotNull String name, @NotNull NodeState after) {
+ return childNodeUpdated(name, EMPTY_NODE, after);
+ }
+
@Override
public boolean childNodeChanged(@NotNull String name, @NotNull NodeState before, @NotNull NodeState after) {
- try {
- SegmentNodeState compacted = compact(before, after, base.getChildNode(name), canceller);
- if (compacted != null) {
- updated();
- builder.setChildNode(name, compacted);
- return true;
- } else {
- return false;
- }
- } catch (IOException e) {
- exception = e;
- return false;
- }
+ return childNodeUpdated(name, before, after);
}
@Override
@@ -237,8 +237,7 @@ public class ClassicCompactor implements Compactor {
}
}
- @NotNull
- protected PropertyState compact(@NotNull PropertyState property) {
+ protected @NotNull PropertyState compact(@NotNull PropertyState property) {
compactionMonitor.onProperty();
String name = property.getName();
Type<?> type = property.getType();
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/Compactor.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/Compactor.java
index ebc52ffdd2..e0dc1037b6 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/Compactor.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/Compactor.java
@@ -19,13 +19,72 @@
package org.apache.jackrabbit.oak.segment;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
import java.io.IOException;
-public interface Compactor {
- SegmentNodeState compact(@NotNull NodeState before, @NotNull NodeState after, @NotNull NodeState onto,
- Canceller canceller) throws IOException;
-}
\ No newline at end of file
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+
+public abstract class Compactor {
+ public final @Nullable CompactedNodeState compactDown(
+ @NotNull NodeState state,
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
+ ) throws IOException {
+ return compactDown(EMPTY_NODE, state, hardCanceller, softCanceller);
+ }
+
+ /**
+ * compact the differences between {@code after} and {@code before} on top of {@code after}.
+ * @param before the node state to diff against from {@code after}
+ * @param after the node state diffed against {@code before}
+ * @param hardCanceller the trigger for hard cancellation, will abandon compaction if cancelled
+ * @param softCanceller the trigger for soft cancellation, will return partially compacted state if cancelled
+ * @return the compacted node state or {@code null} if hard-cancelled
+ * @throws IOException will throw exception if any errors occur during compaction
+ */
+ public abstract @Nullable CompactedNodeState compactDown(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
+ ) throws IOException;
+
+ public final @Nullable CompactedNodeState compactUp(
+ @NotNull NodeState state,
+ @NotNull Canceller canceller
+ ) throws IOException {
+ return compactUp(EMPTY_NODE, state, canceller);
+ }
+
+ /**
+ * compact the differences between {@code after} and {@code before} on top of {@code before}.
+ */
+ public final @Nullable CompactedNodeState compactUp(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull Canceller canceller
+ ) throws IOException {
+ return compact(before, after, before, canceller);
+ }
+
+ /**
+ * compact the differences between {@code after} and {@code before} on top of {@code onto}.
+ * @param before the node state to diff against from {@code after}
+ * @param after the node state diffed against {@code before}
+ * @param onto the node state to compact to apply the diff to
+ * @param canceller the trigger for hard cancellation, will abandon compaction if cancelled
+ * @return the compacted node state or {@code null} if hard-cancelled
+ * @throws IOException will throw exception if any errors occur during compaction
+ */
+ public abstract @Nullable CompactedNodeState compact(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull NodeState onto,
+ @NotNull Canceller canceller
+ ) throws IOException;
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CompactorUtils.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CompactorUtils.java
index ef0bae81c7..51607437c9 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CompactorUtils.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CompactorUtils.java
@@ -24,8 +24,7 @@ import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
final class CompactorUtils {
- @Nullable
- static Buffer getStableIdBytes(@NotNull NodeState state) {
+ static @Nullable Buffer getStableIdBytes(@NotNull NodeState state) {
if (state instanceof SegmentNodeState) {
return ((SegmentNodeState) state).getStableIdBytes();
} else {
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java
index 9a90be2663..3830eb7584 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java
@@ -19,11 +19,14 @@
package org.apache.jackrabbit.oak.segment;
import org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.commons.Buffer;
import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
-import org.apache.jackrabbit.oak.spi.blob.BlobStore;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
import org.jetbrains.annotations.NotNull;
@@ -31,9 +34,11 @@ import org.jetbrains.annotations.Nullable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
-import java.util.Map;
+import java.util.Map.Entry;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -43,26 +48,25 @@ import java.util.concurrent.TimeUnit;
import static org.apache.jackrabbit.guava.common.base.Preconditions.checkNotNull;
import static org.apache.jackrabbit.guava.common.base.Preconditions.checkState;
import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
-import static org.apache.jackrabbit.oak.segment.CompactorUtils.getStableIdBytes;
/**
* This compactor implementation leverages the tree structure of the repository for concurrent compaction.
- * It explores the tree breadth-first until the target node count is reached. Every node at this depth will be
- * an entry point for asynchronous compaction. After the exploration phase, the main thread will collect
- * these compaction results and write their parents' node state to disk.
+ * It explores the tree breadth-first until the target node count ({@value EXPLORATION_LOWER_LIMIT}) is reached.
+ * Every node at this depth will be an entry point for asynchronous compaction. After the exploration phase,
+ * the main thread will collect these compaction results and write their parents' node state to disk.
*/
public class ParallelCompactor extends CheckpointCompactor {
/**
* Expand repository tree until there are this many nodes for each worker to compact. Tradeoff
- * between low efficiency of many small tasks and high risk of at least one of the subtrees being
- * significantly larger than totalSize / numWorkers (unequal work distribution).
+ * between inefficiency of many small tasks and high risk of at least one of the subtrees being
+ * significantly larger than totalSize / {@code numWorkers} (unequal work distribution).
*/
- private static final int MIN_NODES_PER_WORKER = 1000;
+ private static final int EXPLORATION_LOWER_LIMIT = 10_000;
/**
- * Stop expansion if tree size grows beyond this many nodes per worker at the latest.
+ * Stop expansion if tree size grows beyond this many nodes.
*/
- private static final int MAX_NODES_PER_WORKER = 10_000;
+ private static final int EXPLORATION_UPPER_LIMIT = 100_000;
private final int numWorkers;
@@ -71,71 +75,47 @@ public class ParallelCompactor extends CheckpointCompactor {
/**
* Manages workers for asynchronous compaction.
*/
- @Nullable
- private ExecutorService executorService;
+ private @Nullable ExecutorService executorService;
/**
* Create a new instance based on the passed arguments.
- * @param gcListener listener receiving notifications about the garbage collection process
- * @param reader segment reader used to read from the segments
- * @param writer segment writer used to serialise to segments
- * @param blobStore the blob store or {@code null} if none
- * @param compactionMonitor notification call back for each compacted nodes, properties, and binaries
- * @param nThreads number of threads to use for parallel compaction,
- * negative numbers are interpreted relative to the number of available processors
+ *
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @param writer segment writer used to serialise to segments
+ * @param compactionMonitor notification call back for each compacted nodes, properties, and binaries
+ * @param nThreads number of threads to use for parallel compaction,
+ * negative numbers are interpreted relative to the number of available processors
*/
public ParallelCompactor(
@NotNull GCMonitor gcListener,
- @NotNull SegmentReader reader,
- @NotNull SegmentWriter writer,
- @Nullable BlobStore blobStore,
+ @NotNull CompactionWriter writer,
@NotNull GCNodeWriteMonitor compactionMonitor,
int nThreads) {
- super(gcListener, reader, writer, blobStore, compactionMonitor);
-
- int availableProcessors = Runtime.getRuntime().availableProcessors();
+ super(gcListener, writer, compactionMonitor);
if (nThreads < 0) {
- nThreads += availableProcessors + 1;
+ nThreads += Runtime.getRuntime().availableProcessors() + 1;
}
numWorkers = Math.max(0, nThreads - 1);
totalSizeEstimate = compactionMonitor.getEstimatedTotal();
}
/**
- * Calculates the minimum number of entry points for asynchronous compaction.
- */
- private int getMinNodeCount() {
- return numWorkers * MIN_NODES_PER_WORKER;
- }
-
- private int getMaxNodeCount() {
- return numWorkers * MAX_NODES_PER_WORKER;
- }
-
- /**
- * Represents structure of repository changes. Tree is built by exploration process and subsequently
- * used to collect and merge asynchronous compaction results.
+ * Implementation of {@link NodeStateDiff} to represent structure of repository changes.
+ * Tree is built by exploration process and subsequently used to collect and merge
+ * asynchronous compaction results.
*/
private class CompactionTree implements NodeStateDiff {
- @NotNull
- private final NodeState before;
- @NotNull
- private final NodeState after;
- @NotNull
- private final NodeState onto;
- @NotNull
- private final HashMap<String, CompactionTree> modifiedChildren = new HashMap<>();
- @NotNull
- private final List<Property> modifiedProperties = new ArrayList<>();
- @NotNull
- private final List<String> removedChildNames = new ArrayList<>();
- @NotNull
- private final List<String> removedPropertyNames = new ArrayList<>();
+ private final @NotNull NodeState before;
+ private final @NotNull NodeState after;
+ private final @NotNull NodeState onto;
+ private final @NotNull List<Entry<String, CompactionTree>> modifiedChildren = new ArrayList<>();
+ private final @NotNull List<Property> modifiedProperties = new ArrayList<>();
+ private final @NotNull List<String> removedChildNames = new ArrayList<>();
+ private final @NotNull List<String> removedPropertyNames = new ArrayList<>();
/**
* Stores result of asynchronous compaction.
*/
- @Nullable
- private Future<SegmentNodeState> compactionFuture;
+ private @Nullable Future<CompactedNodeState> compactionFuture;
CompactionTree(@NotNull NodeState before, @NotNull NodeState after, @NotNull NodeState onto) {
this.before = checkNotNull(before);
@@ -144,24 +124,35 @@ public class ParallelCompactor extends CheckpointCompactor {
}
private class Property {
- @NotNull
- private final PropertyState state;
+ private final @NotNull PropertyState state;
Property(@NotNull PropertyState state) {
this.state = state;
}
- @NotNull
- PropertyState compact() {
+ @NotNull PropertyState compact() {
return compactor.compact(state);
}
}
- boolean compareStates(Canceller canceller) {
+ private boolean compareState(@NotNull Canceller canceller) {
return after.compareAgainstBaseState(before,
new CancelableDiff(this, () -> canceller.check().isCancelled()));
}
+ @Nullable List<Entry<String, CompactionTree>> expand(@NotNull Canceller hardCanceller) {
+ checkState(compactionFuture == null);
+ CompactedNodeState compactedState = compactor.getPreviouslyCompactedState(after);
+ if (compactedState != null) {
+ compactionFuture = CompletableFuture.completedFuture(compactedState);
+ return Collections.emptyList();
+ } else if (compareState(hardCanceller)) {
+ return modifiedChildren;
+ } else {
+ return null;
+ }
+ }
+
long getEstimatedSize() {
return ApproximateCounter.getCountSync(after);
}
@@ -186,15 +177,17 @@ public class ParallelCompactor extends CheckpointCompactor {
@Override
public boolean childNodeAdded(String name, NodeState after) {
- CompactionTree child = new CompactionTree(EMPTY_NODE, after, EMPTY_NODE);
- modifiedChildren.put(name, child);
+ NodeState childOnto = onto.getChildNode(name);
+ CompactionTree child = new CompactionTree(EMPTY_NODE, after,
+ childOnto.exists() ? childOnto : EMPTY_NODE);
+ modifiedChildren.add(new SimpleImmutableEntry<>(name, child));
return true;
}
@Override
public boolean childNodeChanged(String name, NodeState before, NodeState after) {
CompactionTree child = new CompactionTree(before, after, onto.getChildNode(name));
- modifiedChildren.put(name, child);
+ modifiedChildren.add(new SimpleImmutableEntry<>(name, child));
return true;
}
@@ -207,20 +200,38 @@ public class ParallelCompactor extends CheckpointCompactor {
/**
* Start asynchronous compaction.
*/
- boolean compactAsync(Canceller canceller) {
- if (compactionFuture != null) {
+ void compactAsync(@NotNull Canceller hardCanceller, @Nullable Canceller softCanceller) {
+ if (compactionFuture == null) {
+ checkNotNull(executorService);
+ if (softCanceller == null) {
+ compactionFuture = executorService.submit(() ->
+ compactor.compact(before, after, onto, hardCanceller));
+ } else {
+ checkState(onto.equals(after));
+ compactionFuture = executorService.submit(() ->
+ compactor.compactDown(before, after, hardCanceller, softCanceller));
+ }
+ }
+ }
+
+ /**
+ * Will attempt to cancel pending asynchronous compaction. Already running tasks will not be affected.
+ * Waiting for the compactor to return internally after checking Canceller for all scheduled tasks
+ * causes a lot of overhead which can hereby be avoided.
+ */
+ private boolean tryCancelCompaction() {
+ if (compactionFuture != null && compactionFuture.cancel(false)) {
+ compactionFuture = null;
+ return true;
+ } else {
return false;
}
- checkNotNull(executorService);
- compactionFuture = executorService.submit(() -> compactor.compact(before, after, onto, canceller));
- return true;
}
/**
* Start synchronous compaction on tree or collect result of asynchronous compaction if it has been started.
*/
- @Nullable
- SegmentNodeState compact() throws IOException {
+ @Nullable CompactedNodeState compact() throws IOException {
if (compactionFuture != null) {
try {
return compactionFuture.get();
@@ -231,76 +242,84 @@ public class ParallelCompactor extends CheckpointCompactor {
}
}
- MemoryNodeBuilder builder = new MemoryNodeBuilder(onto);
+ NodeBuilder builder = new MemoryNodeBuilder(onto);
+ Buffer stableIdBytes = CompactorUtils.getStableIdBytes(after);
- for (Map.Entry<String, CompactionTree> entry : modifiedChildren.entrySet()) {
- SegmentNodeState compactedState = entry.getValue().compact();
+ for (int i = 0; i < modifiedChildren.size(); i++) {
+ Entry<String, CompactionTree> entry = modifiedChildren.get(i);
+ CompactionTree child = entry.getValue();
+ CompactedNodeState compactedState = child.compact();
if (compactedState == null) {
return null;
}
builder.setChildNode(entry.getKey(), compactedState);
+
+ // collect results and cancel unfinished tasks in reverse order
+ // increases cancellation success rate since tasks are executed in order
+ if (!compactedState.isComplete()) {
+ for (int j = modifiedChildren.size()-1; j > i; j--) {
+ entry = modifiedChildren.get(j);
+ if (!entry.getValue().tryCancelCompaction()) {
+ compactedState = entry.getValue().compact();
+ if (compactedState == null) {
+ return null;
+ }
+ builder.setChildNode(entry.getKey(), compactedState);
+ }
+ }
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, false);
+ }
}
- for (String childName : removedChildNames) {
- builder.getChildNode(childName).remove();
+
+ for (String name : removedChildNames) {
+ builder.getChildNode(name).remove();
}
+
for (Property property : modifiedProperties) {
builder.setProperty(property.compact());
}
- for (String propertyName : removedPropertyNames) {
- builder.removeProperty(propertyName);
+
+ for (String name : removedPropertyNames) {
+ builder.removeProperty(name);
}
- return compactor.writeNodeState(builder.getNodeState(), getStableIdBytes(after));
+
+ return compactor.writeNodeState(builder.getNodeState(), stableIdBytes, true);
}
}
/**
- * Implementation of {@link NodeStateDiff} to build {@link CompactionTree} and start asynchronous compaction on
+ * Handler class to build {@link CompactionTree} and start asynchronous compaction at
* suitable entry points. Performs what is referred to as the exploration phase in other comments.
*/
private class CompactionHandler {
- @NotNull
- private final NodeState base;
+ private final @NotNull NodeState base;
+ private final @NotNull Canceller hardCanceller;
+ private final @Nullable Canceller softCanceller;
- @NotNull
- private final Canceller canceller;
+ CompactionHandler(@NotNull NodeState base, @NotNull Canceller hardCanceller) {
+ this.base = base;
+ this.hardCanceller = hardCanceller;
+ this.softCanceller = null;
+ }
- CompactionHandler(@NotNull NodeState base, @NotNull Canceller canceller) {
+ CompactionHandler(@NotNull NodeState base, @NotNull Canceller hardCanceller, @NotNull Canceller softCanceller) {
this.base = base;
- this.canceller = canceller;
+ this.hardCanceller = hardCanceller;
+ this.softCanceller = softCanceller;
}
- @Nullable
- SegmentNodeState diff(@NotNull NodeState before, @NotNull NodeState after) throws IOException {
+ @Nullable CompactedNodeState diff(@NotNull NodeState before, @NotNull NodeState after) throws IOException {
checkNotNull(executorService);
checkState(!executorService.isShutdown());
gcListener.info("compacting with {} threads.", numWorkers + 1);
gcListener.info("exploring content tree to find subtrees for parallel compaction.");
- gcListener.info("target node count for expansion is {}, based on {} available workers.",
- getMinNodeCount(), numWorkers);
+ gcListener.info("target node count for expansion is {}.", EXPLORATION_LOWER_LIMIT);
- CompactionTree compactionTree = new CompactionTree(before, after, base);
- if (!compactionTree.compareStates(canceller)) {
- return null;
- }
-
- List<CompactionTree> topLevel = new ArrayList<>();
- for (Map.Entry<String, CompactionTree> childEntry : compactionTree.modifiedChildren.entrySet()) {
- switch (childEntry.getKey()) {
- // these tend to be the largest directories, others will not be split up
- case "content":
- case "oak:index":
- case "jcr:system":
- topLevel.add(childEntry.getValue());
- break;
- default:
- checkState(childEntry.getValue().compactAsync(canceller));
- break;
- }
- }
+ CompactionTree root = new CompactionTree(before, after, base);
- if (diff(1, topLevel)) {
- SegmentNodeState compacted = compactionTree.compact();
+ if (diff(0, Collections.singletonList(root))) {
+ CompactedNodeState compacted = root.compact();
if (compacted != null) {
return compacted;
}
@@ -320,14 +339,12 @@ public class ParallelCompactor extends CheckpointCompactor {
}
private boolean diff(int depth, List<CompactionTree> nodes) {
- int targetCount = getMinNodeCount();
- gcListener.info("Found {} nodes at depth {}, target is {}.", nodes.size(), depth, targetCount);
+ gcListener.info("found {} nodes at depth {}.", nodes.size(), depth);
- if (nodes.size() >= targetCount) {
- nodes.forEach(node -> node.compactAsync(canceller));
+ if (nodes.size() >= EXPLORATION_LOWER_LIMIT) {
+ nodes.forEach(node -> node.compactAsync(hardCanceller, softCanceller));
return true;
} else if (nodes.isEmpty()) {
- gcListener.info("Amount of changes too small, tree will not be split.");
return true;
}
@@ -335,35 +352,63 @@ public class ParallelCompactor extends CheckpointCompactor {
for (CompactionTree node : nodes) {
long estimatedSize = node.getEstimatedSize();
if (estimatedSize != -1 && estimatedSize <= (totalSizeEstimate / numWorkers)) {
- checkState(node.compactAsync(canceller));
- } else if (nextDepth.size() < getMaxNodeCount()) {
- if (!node.compareStates(canceller)) {
+ node.compactAsync(hardCanceller, softCanceller);
+ } else if (nextDepth.size() < EXPLORATION_UPPER_LIMIT) {
+ List<Entry<String, CompactionTree>> children = node.expand(hardCanceller);
+ if (children == null) {
return false;
}
- nextDepth.addAll(node.modifiedChildren.values());
+ children.forEach(entry -> nextDepth.add(entry.getValue()));
} else {
nextDepth.add(node);
}
}
+ if (nextDepth.size() < nodes.size()) {
+ nodes.forEach(node -> node.compactAsync(hardCanceller, softCanceller));
+ return true;
+ }
+
return diff(depth + 1, nextDepth);
}
}
- @Nullable
+ private boolean initializeExecutor() {
+ if (numWorkers <= 0) {
+ gcListener.info("using sequential compaction.");
+ return false;
+ }
+ if (executorService == null || executorService.isShutdown()) {
+ executorService = Executors.newFixedThreadPool(numWorkers);
+ }
+ return true;
+ }
+
+ @Override
+ protected @Nullable CompactedNodeState compactDownWithDelegate(
+ @NotNull NodeState before,
+ @NotNull NodeState after,
+ @NotNull Canceller hardCanceller,
+ @NotNull Canceller softCanceller
+ ) throws IOException {
+ if (initializeExecutor()) {
+ return new CompactionHandler(after, hardCanceller, softCanceller).diff(before, after);
+ } else {
+ return super.compactDownWithDelegate(before, after, hardCanceller, softCanceller);
+ }
+ }
+
@Override
- protected SegmentNodeState compactWithDelegate(
+ protected @Nullable CompactedNodeState compactWithDelegate(
@NotNull NodeState before,
@NotNull NodeState after,
@NotNull NodeState onto,
- Canceller canceller
+ @NotNull Canceller canceller
) throws IOException {
- if (numWorkers <= 0) {
- gcListener.info("using sequential compaction.");
+ if (initializeExecutor()) {
+ return new CompactionHandler(onto, canceller).diff(before, after);
+ } else {
return super.compactWithDelegate(before, after, onto, canceller);
- } else if (executorService == null || executorService.isShutdown()) {
- executorService = Executors.newFixedThreadPool(numWorkers);
}
- return new CompactionHandler(onto, canceller).diff(before, after);
}
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/SegmentWriterFactory.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriterFactory.java
similarity index 87%
rename from oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/SegmentWriterFactory.java
rename to oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriterFactory.java
index 572f7a8557..1eef291c80 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/SegmentWriterFactory.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriterFactory.java
@@ -17,13 +17,10 @@
* under the License.
*/
-package org.apache.jackrabbit.oak.segment.file;
+package org.apache.jackrabbit.oak.segment;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
-interface SegmentWriterFactory {
-
+public interface SegmentWriterFactory {
SegmentWriter newSegmentWriter(GCGeneration generation);
-
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractCompactionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractCompactionStrategy.java
index 4851994548..f368f8ec2a 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractCompactionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractCompactionStrategy.java
@@ -27,8 +27,6 @@ import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCStatus.COMPA
import static org.apache.jackrabbit.oak.segment.file.TarRevisions.EXPEDITE_OPTION;
import static org.apache.jackrabbit.oak.segment.file.TarRevisions.timeout;
-import org.apache.jackrabbit.guava.common.base.Function;
-
import org.apache.jackrabbit.oak.segment.CheckpointCompactor;
import org.apache.jackrabbit.oak.segment.ClassicCompactor;
import org.apache.jackrabbit.oak.segment.Compactor;
@@ -44,12 +42,15 @@ import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
abstract class AbstractCompactionStrategy implements CompactionStrategy {
abstract GCType getCompactionType();
- abstract GCGeneration nextGeneration(GCGeneration current);
+ abstract GCGeneration partialGeneration(GCGeneration current);
+
+ abstract GCGeneration targetGeneration(GCGeneration current);
private CompactionResult compactionSucceeded(
Context context,
@@ -60,6 +61,20 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
return CompactionResult.succeeded(getCompactionType(), generation, context.getGCOptions(), compactedRootId, context.getGCCount());
}
+ private CompactionResult compactionPartiallySucceeded(
+ Context context,
+ GCGeneration generation,
+ RecordId compactedRootId
+ ) {
+ context.getGCListener().compactionSucceeded(generation);
+ return CompactionResult.partiallySucceeded(generation, compactedRootId, context.getGCCount());
+ }
+
+ private static CompactionResult compactionAborted(Context context, GCGeneration generation) {
+ context.getGCListener().compactionFailed(generation);
+ return CompactionResult.aborted(getGcGeneration(context), context.getGCCount());
+ }
+
private static GCGeneration getGcGeneration(Context context) {
return context.getRevisions().getHead().getSegmentId().getGcGeneration();
}
@@ -72,23 +87,21 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
return context.getTarFiles().size();
}
- private static CompactionResult compactionAborted(Context context, GCGeneration generation) {
- context.getGCListener().compactionFailed(generation);
- return CompactionResult.aborted(getGcGeneration(context), generation, context.getGCCount());
- }
-
- private static SegmentNodeState forceCompact(
+ private static CompactedNodeState forceCompact(
Context context,
NodeState base,
NodeState onto,
Compactor compactor,
Canceller canceller
) throws InterruptedException {
- RecordId compactedId = setHead(context, headId -> {
+ AtomicReference<CompactedNodeState> compacted = new AtomicReference<>();
+ context.getRevisions().setHead(headId -> {
try {
PrintableStopwatch t = PrintableStopwatch.createStarted();
- SegmentNodeState after = compactor.compact(base, context.getSegmentReader().readNode(headId), onto, canceller);
+ NodeState currentHead = context.getSegmentReader().readNode(headId);
+ CompactedNodeState after = compactor.compact(base, currentHead, onto, canceller);
if (after != null) {
+ compacted.set(after);
return after.getRecordId();
}
context.getGCListener().info("compaction cancelled after {}", t);
@@ -97,15 +110,8 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
context.getGCListener().error("error during forced compaction.", e);
return null;
}
- });
- if (compactedId == null) {
- return null;
- }
- return context.getSegmentReader().readNode(compactedId);
- }
-
- private static RecordId setHead(Context context, Function<RecordId, RecordId> f) throws InterruptedException {
- return context.getRevisions().setHead(f, timeout(context.getGCOptions().getForceTimeout(), SECONDS));
+ }, timeout(context.getGCOptions().getForceTimeout(), SECONDS));
+ return compacted.get();
}
private static String formatCompactionType(GCType compactionType) {
@@ -119,68 +125,109 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
}
}
+ private boolean setHead(Context context, SegmentNodeState previous, SegmentNodeState head) {
+ return context.getRevisions().setHead(previous.getRecordId(), head.getRecordId(), EXPEDITE_OPTION);
+ }
+
final CompactionResult compact(Context context, NodeState base) {
context.getGCListener().info("running {} compaction", formatCompactionType(getCompactionType()));
- GCGeneration nextGeneration = nextGeneration(getGcGeneration(context));
+ GCGeneration baseGeneration = getGcGeneration(context);
+ GCGeneration partialGeneration = partialGeneration(baseGeneration);
+ GCGeneration targetGeneration = targetGeneration(baseGeneration);
+ GCIncrement gcIncrement = new GCIncrement(baseGeneration, partialGeneration, targetGeneration);
try {
PrintableStopwatch watch = PrintableStopwatch.createStarted();
context.getGCListener().info(
- "compaction started, gc options={}, current generation={}, new generation={}",
+ "compaction started, gc options={},\n{}",
context.getGCOptions(),
- getHead(context).getRecordId().getSegment().getGcGeneration(),
- nextGeneration
+ gcIncrement
);
context.getGCListener().updateStatus(COMPACTION.message());
GCJournal.GCJournalEntry gcEntry = context.getGCJournal().read();
long initialSize = size(context);
- SegmentWriter writer = context.getSegmentWriterFactory().newSegmentWriter(nextGeneration);
+ CompactionWriter writer = new CompactionWriter(
+ context.getSegmentReader(),
+ context.getBlobStore(),
+ gcIncrement,
+ context.getSegmentWriterFactory());
context.getCompactionMonitor().init(gcEntry.getRepoSize(), gcEntry.getNodes(), initialSize);
- Canceller compactionCanceller = context.getCanceller().withShortCircuit();
+ Canceller hardCanceller = context.getHardCanceller().withShortCircuit();
+ Canceller softCanceller = context.getSoftCanceller().withShortCircuit();
Compactor compactor = newCompactor(context, writer);
+ CompactedNodeState compacted = null;
- SegmentNodeState head = getHead(context);
- SegmentNodeState compacted = compactor.compact(base, head, base, compactionCanceller);
- if (compacted == null) {
- context.getGCListener().warn("compaction cancelled: {}.", compactionCanceller.check().getReason().orElse("unknown reason"));
- return compactionAborted(context, nextGeneration);
- }
+ int cycles;
+ boolean success;
+ final int retryCount = Math.max(0, context.getGCOptions().getRetryCount());
- context.getGCListener().info("compaction cycle 0 completed in {}. Compacted {} to {}",
- watch, head.getRecordId(), compacted.getRecordId());
-
- int cycles = 0;
- boolean success = false;
- SegmentNodeState previousHead = head;
- while (cycles < context.getGCOptions().getRetryCount() &&
- !(success = context.getRevisions().setHead(previousHead.getRecordId(), compacted.getRecordId(), EXPEDITE_OPTION))) {
- // Some other concurrent changes have been made.
- // Rebase (and compact) those changes on top of the
- // compacted state before retrying to set the head.
- cycles++;
- context.getGCListener().info("compaction detected concurrent commits while compacting. " +
- "Compacting these commits. Cycle {} of {}",
- cycles, context.getGCOptions().getRetryCount());
- context.getGCListener().updateStatus(COMPACTION_RETRY.message() + cycles);
- PrintableStopwatch cycleWatch = PrintableStopwatch.createStarted();
+ SegmentNodeState head;
+ Flusher flusher = () -> {
+ writer.flush();
+ context.getFlusher().flush();
+ };
+ do {
head = getHead(context);
- compacted = compactor.compact(previousHead, head, compacted, compactionCanceller);
+ SegmentNodeState after = (compacted == null) ? head : compacted;
+ Canceller stateSaveTrigger = context.getStateSaveTriggerSupplier().get().withShortCircuit();
+
+ if (stateSaveTrigger.isCancelable()) {
+ context.getGCListener().info("intermediate state save enabled.");
+ Canceller saveStateCanceller = softCanceller.withCondition(
+ "save intermediate compaction state", () -> stateSaveTrigger.check().isCancelled());
+ compacted = compactor.compactDown(base, after, hardCanceller, saveStateCanceller);
+ } else if (softCanceller.isCancelable()) {
+ context.getGCListener().info("soft cancellation enabled.");
+ compacted = compactor.compactDown(base, after, hardCanceller, softCanceller);
+ } else {
+ compacted = compactor.compactUp(base, after, hardCanceller);
+ }
+
if (compacted == null) {
- context.getGCListener().warn("compaction cancelled: {}.", compactionCanceller.check().getReason().orElse("unknown reason"));
- return compactionAborted(context, nextGeneration);
+ context.getGCListener().warn("compaction cancelled: {}.",
+ hardCanceller.check().getReason().orElse("unknown reason"));
+ return compactionAborted(context, targetGeneration);
}
- context.getGCListener().info("compaction cycle {} completed in {}. Compacted {} against {} to {}",
- cycles, cycleWatch, head.getRecordId(), previousHead.getRecordId(), compacted.getRecordId());
- previousHead = head;
- }
+ context.getGCListener().info("compaction cycle 0 completed in {}. Compacted {} to {}",
+ watch, head.getRecordId(), compacted.getRecordId());
+
+ cycles = 0;
+
+ while (!(success = setHead(context, head, compacted)) && cycles < retryCount) {
+ // Some other concurrent changes have been made.
+ // Rebase (and compact) those changes on top of the
+ // compacted state before retrying to set the head.
+ cycles++;
+ context.getGCListener().info("compaction detected concurrent commits while compacting. " +
+ "Compacting these commits. Cycle {} of {}", cycles, retryCount);
+ context.getGCListener().updateStatus(COMPACTION_RETRY.message() + cycles);
+ PrintableStopwatch cycleWatch = PrintableStopwatch.createStarted();
+
+ SegmentNodeState newHead = getHead(context);
+ compacted = compactor.compact(head, newHead,compacted, hardCanceller);
+ if (compacted == null) {
+ context.getGCListener().warn("compaction cancelled: {}.",
+ hardCanceller.check().getReason().orElse("unknown reason"));
+ return compactionAborted(context, targetGeneration);
+ }
+
+ context.getGCListener().info("compaction cycle {} completed in {}. Compacted {} against {} to {}",
+ cycles, cycleWatch, head.getRecordId(), newHead.getRecordId(), compacted.getRecordId());
+ head = newHead;
+ }
+
+ if (success) {
+ flusher.flush();
+ }
+ } while (success && !compacted.isComplete() && !softCanceller.check().isCancelled());
if (!success) {
context.getGCListener().info("compaction gave up compacting concurrent commits after {} cycles.", cycles);
@@ -194,12 +241,13 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
cycles++;
- Canceller forcedCompactionCanceller = compactionCanceller
+ Canceller forcedCompactionCanceller = hardCanceller
.withTimeout("forced compaction timeout exceeded", forceTimeout, SECONDS)
.withShortCircuit();
- compacted = forceCompact(context, previousHead, compacted, compactor, forcedCompactionCanceller);
- success = compacted != null;
- if (success) {
+ compacted = forceCompact(context, head, compacted, compactor, forcedCompactionCanceller);
+ if (compacted != null) {
+ success = true;
+ flusher.flush();
context.getGCListener().info("compaction succeeded to force compact remaining commits after {}.", forceWatch);
} else {
Cancellation cancellation = forcedCompactionCanceller.check();
@@ -219,38 +267,40 @@ abstract class AbstractCompactionStrategy implements CompactionStrategy {
if (success) {
// Update type of the last compaction before calling methods that could throw an exception.
context.getSuccessfulCompactionListener().onSuccessfulCompaction(getCompactionType());
- writer.flush();
- context.getFlusher().flush();
- context.getGCListener().info("compaction succeeded in {}, after {} cycles", watch, cycles);
context.getCompactionMonitor().finished();
- return compactionSucceeded(context, nextGeneration, compacted.getRecordId());
+
+ if (compacted.isComplete()) {
+ context.getGCListener().info("compaction succeeded in {}, after {} cycles", watch, cycles);
+ return compactionSucceeded(context, targetGeneration, compacted.getRecordId());
+ } else {
+ context.getGCListener().info("compaction partially succeeded in {}: {}.",
+ watch, softCanceller.check().getReason().orElse("unknown reason"));
+ return compactionPartiallySucceeded(context, partialGeneration, compacted.getRecordId());
+ }
} else {
context.getGCListener().info("compaction failed after {}, and {} cycles", watch, cycles);
- return compactionAborted(context, nextGeneration);
+ return compactionAborted(context, targetGeneration);
}
} catch (InterruptedException e) {
context.getGCListener().error("compaction interrupted", e);
currentThread().interrupt();
- return compactionAborted(context, nextGeneration);
+ return compactionAborted(context, targetGeneration);
} catch (Throwable e) {
context.getGCListener().error("compaction encountered an error", e instanceof Exception ? (Exception) e : new Exception(e));
- return compactionAborted(context, nextGeneration);
+ return compactionAborted(context, targetGeneration);
}
}
- private Compactor newCompactor(Context context, SegmentWriter writer) {
+ private Compactor newCompactor(Context context, CompactionWriter writer) {
CompactorType compactorType = context.getGCOptions().getCompactorType();
switch (compactorType) {
case PARALLEL_COMPACTOR:
- return new ParallelCompactor(context.getGCListener(), context.getSegmentReader(), writer,
- context.getBlobStore(), context.getCompactionMonitor(),
+ return new ParallelCompactor(context.getGCListener(), writer, context.getCompactionMonitor(),
context.getGCOptions().getConcurrency());
case CHECKPOINT_COMPACTOR:
- return new CheckpointCompactor(context.getGCListener(), context.getSegmentReader(), writer,
- context.getBlobStore(), context.getCompactionMonitor());
+ return new CheckpointCompactor(context.getGCListener(), writer, context.getCompactionMonitor());
case CLASSIC_COMPACTOR:
- return new ClassicCompactor(context.getSegmentReader(), writer, context.getBlobStore(),
- context.getCompactionMonitor());
+ return new ClassicCompactor(writer, context.getCompactionMonitor());
default:
throw new IllegalArgumentException("Unknown compactor type: " + compactorType);
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractGarbageCollectionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractGarbageCollectionStrategy.java
index 0f418bedf0..6d219c5f9c 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractGarbageCollectionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractGarbageCollectionStrategy.java
@@ -24,12 +24,14 @@ import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCStatus.IDLE;
import java.io.IOException;
import java.util.List;
+import java.util.function.Supplier;
import org.apache.jackrabbit.guava.common.base.Predicate;
import org.apache.jackrabbit.oak.segment.Revisions;
import org.apache.jackrabbit.oak.segment.SegmentCache;
import org.apache.jackrabbit.oak.segment.SegmentReader;
import org.apache.jackrabbit.oak.segment.SegmentTracker;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
@@ -151,7 +153,7 @@ abstract class AbstractGarbageCollectionStrategy implements GarbageCollectionStr
return context.getRevisions().getHead().getSegmentId().getGcGeneration();
}
- private List<String> cleanup(Context context, CompactionResult compactionResult) throws IOException {
+ public List<String> cleanup(Context context, CompactionResult compactionResult) throws IOException {
return getCleanupStrategy().cleanup(newCleanupStrategyContext(context, compactionResult));
}
@@ -230,10 +232,20 @@ abstract class AbstractGarbageCollectionStrategy implements GarbageCollectionStr
}
@Override
- public Canceller getCanceller() {
+ public Canceller getHardCanceller() {
return context.getCanceller();
}
+ @Override
+ public Canceller getSoftCanceller() {
+ return Canceller.newCanceller();
+ }
+
+ @Override
+ public Supplier<Canceller> getStateSaveTriggerSupplier() {
+ return Canceller::newCanceller;
+ }
+
@Override
public int getGCCount() {
return context.getGCCount();
@@ -282,7 +294,7 @@ abstract class AbstractGarbageCollectionStrategy implements GarbageCollectionStr
@Override
public GCJournal getGCJournal() {
- return context.getGCJournal();
+ return compactionResult.requiresGCJournalEntry() ? context.getGCJournal() : null;
}
@Override
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CleanupFirstCompactionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CleanupFirstCompactionStrategy.java
index 07c5e2c8db..c4150f70e1 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CleanupFirstCompactionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CleanupFirstCompactionStrategy.java
@@ -19,20 +19,13 @@
package org.apache.jackrabbit.oak.segment.file;
-import static org.apache.jackrabbit.guava.common.collect.Sets.newHashSet;
-import static org.apache.jackrabbit.oak.segment.SegmentId.isDataSegmentId;
import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCStatus.CLEANUP;
import static org.apache.jackrabbit.oak.segment.file.PrintableBytes.newPrintableBytes;
import java.io.IOException;
-import java.util.Collection;
import java.util.List;
-import java.util.Set;
-import java.util.UUID;
import org.apache.jackrabbit.guava.common.base.Joiner;
-import org.apache.jackrabbit.guava.common.base.Predicate;
-import org.apache.jackrabbit.oak.segment.SegmentId;
import org.apache.jackrabbit.oak.segment.file.tar.CleanupContext;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
@@ -76,46 +69,6 @@ class CleanupFirstCompactionStrategy implements CompactionStrategy {
PrintableStopwatch watch = PrintableStopwatch.createStarted();
- Predicate<GCGeneration> reclaimer;
-
- GCGeneration currentGeneration = context.getRevisions().getHead().getSegmentId().getGcGeneration();
-
- switch (context.getGCOptions().getGCType()) {
- case FULL:
- reclaimer = generation -> {
- if (generation == null) {
- return false;
- }
- if (generation.getFullGeneration() < currentGeneration.getFullGeneration()) {
- return true;
- }
- if (generation.getFullGeneration() > currentGeneration.getFullGeneration()) {
- return true;
- }
- return generation.getGeneration() < currentGeneration.getGeneration() && !generation.isCompacted();
- };
- break;
- case TAIL:
- reclaimer = generation -> {
- if (generation == null) {
- return false;
- }
- if (generation.getFullGeneration() < currentGeneration.getFullGeneration() - 1) {
- return true;
- }
- if (generation.getFullGeneration() == currentGeneration.getFullGeneration() - 1) {
- return !generation.isCompacted();
- }
- if (generation.getFullGeneration() > currentGeneration.getFullGeneration()) {
- return true;
- }
- return generation.getGeneration() < currentGeneration.getGeneration() && !generation.isCompacted();
- };
- break;
- default:
- throw new IllegalArgumentException("invalid garbage collection type");
- }
-
context.getGCListener().info("pre-compaction cleanup started");
context.getGCListener().updateStatus(CLEANUP.message());
@@ -124,7 +77,7 @@ class CleanupFirstCompactionStrategy implements CompactionStrategy {
System.gc();
- TarFiles.CleanupResult cleanupResult = context.getTarFiles().cleanup(newCleanupContext(context, reclaimer));
+ TarFiles.CleanupResult cleanupResult = context.getTarFiles().cleanup(newCleanupContext(context));
if (cleanupResult.isInterrupted()) {
context.getGCListener().info("cleanup interrupted");
@@ -149,39 +102,37 @@ class CleanupFirstCompactionStrategy implements CompactionStrategy {
return strategy.compact(context);
}
- private static CleanupContext newCleanupContext(Context context, Predicate<GCGeneration> old) {
- return new CleanupContext() {
-
- private boolean isUnreferencedBulkSegment(UUID id, boolean referenced) {
- return !isDataSegmentId(id.getLeastSignificantBits()) && !referenced;
- }
-
- private boolean isOldDataSegment(UUID id, GCGeneration generation) {
- return isDataSegmentId(id.getLeastSignificantBits()) && old.apply(generation);
- }
+ private static CleanupContext newCleanupContext(Context context) {
+ GCGeneration currentGeneration = context.getRevisions().getHead().getSegmentId().getGcGeneration();
+ String compactedRoot = context.getGCJournal().read().getRoot();
- @Override
- public Collection<UUID> initialReferences() {
- Set<UUID> references = newHashSet();
- for (SegmentId id : context.getSegmentTracker().getReferencedSegmentIds()) {
- if (id.isBulkSegmentId()) {
- references.add(id.asUUID());
+ switch (context.getGCOptions().getGCType()) {
+ case FULL:
+ return new DefaultCleanupContext(context.getSegmentTracker(), generation -> {
+ if (generation == null) {
+ return false;
}
- }
- return references;
- }
-
- @Override
- public boolean shouldReclaim(UUID id, GCGeneration generation, boolean referenced) {
- return isUnreferencedBulkSegment(id, referenced) || isOldDataSegment(id, generation);
- }
-
- @Override
- public boolean shouldFollow(UUID from, UUID to) {
- return !isDataSegmentId(to.getLeastSignificantBits());
- }
-
- };
+ if (generation.getFullGeneration() < currentGeneration.getFullGeneration()) {
+ return true;
+ }
+ return generation.getGeneration() < currentGeneration.getGeneration() && !generation.isCompacted();
+ }, compactedRoot);
+ case TAIL:
+ return new DefaultCleanupContext(context.getSegmentTracker(), generation -> {
+ if (generation == null) {
+ return false;
+ }
+ if (generation.getFullGeneration() < currentGeneration.getFullGeneration() - 1) {
+ return true;
+ }
+ if (generation.getFullGeneration() == currentGeneration.getFullGeneration() - 1) {
+ return !generation.isCompacted();
+ }
+ return generation.getGeneration() < currentGeneration.getGeneration() && !generation.isCompacted();
+ }, compactedRoot);
+ default:
+ throw new IllegalArgumentException("invalid garbage collection type");
+ }
}
private static String toFileNames(@NotNull List<String> files) {
@@ -191,13 +142,4 @@ class CleanupFirstCompactionStrategy implements CompactionStrategy {
return Joiner.on(",").join(files);
}
}
-
- private static GCGeneration getGcGeneration(Context context) {
- return context.getRevisions().getHead().getSegmentId().getGcGeneration();
- }
-
- private static long size(Context context) {
- return context.getTarFiles().size();
- }
-
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactedNodeState.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactedNodeState.java
new file mode 100644
index 0000000000..de31888f91
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactedNodeState.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.SegmentNodeState;
+import org.apache.jackrabbit.oak.segment.SegmentReader;
+import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Simple wrapper class for {@code SegmentNodeState} to keep track of fully and partially compacted nodes.
+ */
+public abstract class CompactedNodeState extends SegmentNodeState {
+ final boolean complete;
+
+ private CompactedNodeState(
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull RecordId id,
+ boolean complete) {
+ super(reader, writer, blobStore, id);
+ this.complete = complete;
+ }
+
+ final public boolean isComplete() {
+ return complete;
+ }
+
+ final static class FullyCompactedNodeState extends CompactedNodeState {
+ FullyCompactedNodeState(
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull RecordId id) {
+ super(reader, writer, blobStore, id, true);
+ }
+ }
+
+ final static class PartiallyCompactedNodeState extends CompactedNodeState {
+ PartiallyCompactedNodeState(
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull RecordId id) {
+ super(reader, writer, blobStore, id, false);
+ }
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionResult.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionResult.java
index f2a447b8d9..4de088e86e 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionResult.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionResult.java
@@ -74,6 +74,41 @@ abstract class CompactionResult {
RecordId getCompactedRootId() {
return compactedRootId;
}
+
+ @Override
+ boolean requiresGCJournalEntry() {
+ return true;
+ }
+ };
+ }
+
+ /**
+ * Result of incremental compaction if the compaction cycle is incomplete.
+ *
+ * @param newGeneration the generation successfully created by compaction
+ * @param compactedRootId the record id of the root created by compaction
+ */
+ static CompactionResult partiallySucceeded(
+ @NotNull GCGeneration newGeneration,
+ @NotNull final RecordId compactedRootId,
+ int gcCount
+ ) {
+ return new CompactionResult(newGeneration, gcCount) {
+
+ @Override
+ Predicate<GCGeneration> reclaimer() {
+ return generation -> false;
+ }
+
+ @Override
+ boolean isSuccess() {
+ return true;
+ }
+
+ @Override
+ RecordId getCompactedRootId() {
+ return compactedRootId;
+ }
};
}
@@ -81,19 +116,16 @@ abstract class CompactionResult {
* Result of an aborted compaction.
*
* @param currentGeneration the current generation of the store
- * @param failedGeneration the generation that compaction attempted to
- * create
*/
static CompactionResult aborted(
@NotNull GCGeneration currentGeneration,
- @NotNull final GCGeneration failedGeneration,
int gcCount
) {
return new CompactionResult(currentGeneration, gcCount) {
@Override
Predicate<GCGeneration> reclaimer() {
- return Reclaimers.newExactReclaimer(failedGeneration);
+ return Reclaimers.newEmptyReclaimer();
}
@Override
@@ -181,6 +213,10 @@ abstract class CompactionResult {
return false;
}
+ boolean requiresGCJournalEntry() {
+ return false;
+ }
+
/**
* @return a diagnostic message describing the outcome of this compaction.
*/
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategy.java
index 73109b269a..b7e0784538 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategy.java
@@ -20,10 +20,12 @@
package org.apache.jackrabbit.oak.segment.file;
import java.io.IOException;
+import java.util.function.Supplier;
import org.apache.jackrabbit.oak.segment.Revisions;
import org.apache.jackrabbit.oak.segment.SegmentReader;
import org.apache.jackrabbit.oak.segment.SegmentTracker;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
@@ -51,7 +53,11 @@ interface CompactionStrategy {
BlobStore getBlobStore();
- Canceller getCanceller();
+ Canceller getHardCanceller();
+
+ Canceller getSoftCanceller();
+
+ Supplier<Canceller> getStateSaveTriggerSupplier();
int getGCCount();
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionWriter.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionWriter.java
new file mode 100644
index 0000000000..cd20a1165f
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/CompactionWriter.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.commons.Buffer;
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.SegmentNodeState;
+import org.apache.jackrabbit.oak.segment.SegmentReader;
+import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.jackrabbit.oak.segment.file.CompactedNodeState.PartiallyCompactedNodeState;
+import static org.apache.jackrabbit.oak.segment.file.CompactedNodeState.FullyCompactedNodeState;
+
+/**
+ * The CompactionWriter delegates compaction calls to the correct {@code SegmentWriter} based on GCGeneration.
+ */
+public class CompactionWriter {
+ private final @NotNull SegmentReader reader;
+ private final @Nullable BlobStore blobStore;
+ private final @NotNull GCIncrement gcIncrement;
+ private final @NotNull SegmentWriter partialWriter;
+ private final @NotNull SegmentWriter targetWriter;
+
+ public CompactionWriter(
+ @NotNull SegmentReader reader,
+ @Nullable BlobStore blobStore,
+ @NotNull GCGeneration generation,
+ @NotNull SegmentWriter segmentWriter) {
+ this.reader = reader;
+ this.blobStore = blobStore;
+ this.gcIncrement = new GCIncrement(generation, generation, generation);
+ this.partialWriter = segmentWriter;
+ this.targetWriter = segmentWriter;
+ }
+
+ public CompactionWriter(
+ @NotNull SegmentReader reader,
+ @Nullable BlobStore blobStore,
+ @NotNull GCIncrement gcIncrement,
+ @NotNull SegmentWriterFactory segmentWriterFactory) {
+ this.reader = reader;
+ this.blobStore = blobStore;
+ this.gcIncrement = gcIncrement;
+ this.partialWriter = gcIncrement.createPartialWriter(segmentWriterFactory);
+ this.targetWriter = gcIncrement.createTargetWriter(segmentWriterFactory);
+ }
+
+ public @NotNull FullyCompactedNodeState writeFullyCompactedNode(
+ @NotNull NodeState nodeState,
+ @Nullable Buffer stableId
+ ) throws IOException {
+ RecordId nodeId = targetWriter.writeNode(nodeState, stableId);
+ return new FullyCompactedNodeState(reader, targetWriter, blobStore, nodeId);
+ }
+
+ public @Nullable PartiallyCompactedNodeState writePartiallyCompactedNode(
+ @NotNull NodeState nodeState,
+ @Nullable Buffer stableId
+ ) throws IOException {
+ RecordId nodeId = partialWriter.writeNode(nodeState, stableId);
+ return new PartiallyCompactedNodeState(reader, partialWriter, blobStore, nodeId);
+ }
+
+ public void flush() throws IOException {
+ partialWriter.flush();
+ targetWriter.flush();
+ }
+
+ public @Nullable FullyCompactedNodeState getPreviouslyCompactedState(NodeState nodeState) {
+ if (!(nodeState instanceof SegmentNodeState)) {
+ return null;
+ }
+ SegmentNodeState segmentNodeState = (SegmentNodeState) nodeState;
+ if (!gcIncrement.isFullyCompacted(segmentNodeState.getGcGeneration())) {
+ return null;
+ }
+ RecordId nodeId = segmentNodeState.getRecordId();
+ return new FullyCompactedNodeState(reader, targetWriter, blobStore, nodeId);
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupContext.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupContext.java
new file mode 100644
index 0000000000..388247d903
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupContext.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.SegmentId;
+import org.apache.jackrabbit.oak.segment.SegmentTracker;
+import org.apache.jackrabbit.oak.segment.file.tar.CleanupContext;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.jackrabbit.oak.segment.SegmentId.isDataSegmentId;
+
+class DefaultCleanupContext implements CleanupContext {
+ private final @NotNull SegmentTracker segmentTracker;
+ private final @NotNull Predicate<GCGeneration> old;
+ private final @Nullable UUID rootSegmentUUID;
+ private boolean aheadOfRoot;
+
+ DefaultCleanupContext(@NotNull SegmentTracker tracker, @NotNull Predicate<GCGeneration> old, @NotNull String compactedRoot) {
+ this.segmentTracker = tracker;
+ this.old = old;
+
+ RecordId rootId = RecordId.fromString(tracker, compactedRoot);
+ if (rootId.equals(RecordId.NULL)) {
+ rootSegmentUUID = null;
+ aheadOfRoot = false;
+ } else {
+ rootSegmentUUID = rootId.getSegmentId().asUUID();
+ aheadOfRoot = true;
+ }
+ }
+
+ /**
+ * Reference-based reclamation for bulk segments.
+ */
+ private boolean isUnreferencedBulkSegment(UUID id, boolean referenced) {
+ return !isDataSegmentId(id.getLeastSignificantBits()) && !referenced;
+ }
+
+ /**
+ * Generational reclamation for data segments.
+ */
+ private boolean isOldDataSegment(UUID id, GCGeneration generation) {
+ return isDataSegmentId(id.getLeastSignificantBits()) && old.test(generation);
+ }
+
+ /**
+ * Special reclamation for unused future segments. Aborting compaction will lead to persisted, but unused
+ * TAR entries with higher generation than the root and set compacted flag. Due to incremental compaction,
+ * a purely generational approach for this cleanup is no longer feasible as segments of higher generation
+ * than the root may be part of a valid repository tree. Observation: compacted segments are unused iff
+ * they are persisted after the last compacted root. This context relies on the cleanup algorithm to mark
+ * TAR entries in reverse order and will consider each compacted segment to be reclaimable until the root
+ * has been encountered, i.e. as long as {@code aheadOfRoot} is true.
+ */
+ private boolean isDanglingFutureSegment(UUID id, GCGeneration generation) {
+ return (aheadOfRoot &= !id.equals(rootSegmentUUID)) && generation.isCompacted();
+ }
+
+ /**
+ * Returns IDs of directly referenced segments. Since reference-based reclamation
+ * is only used for bulk segments, data segment IDs are filtered out.
+ */
+ @Override
+ public Set<UUID> initialReferences() {
+ return segmentTracker.getReferencedSegmentIds().stream()
+ .filter(SegmentId::isBulkSegmentId)
+ .map(SegmentId::asUUID)
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public boolean shouldReclaim(UUID id, GCGeneration generation, boolean referenced) {
+ return isDanglingFutureSegment(id, generation) || isUnreferencedBulkSegment(id, referenced) ||
+ isOldDataSegment(id, generation);
+ }
+
+ @Override
+ public boolean shouldFollow(UUID from, UUID to) {
+ return !isDataSegmentId(to.getLeastSignificantBits());
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupStrategy.java
index 9e202a2cd4..8cb636d846 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/DefaultCleanupStrategy.java
@@ -19,20 +19,14 @@
package org.apache.jackrabbit.oak.segment.file;
-import static org.apache.jackrabbit.guava.common.collect.Sets.newHashSet;
-import static org.apache.jackrabbit.oak.segment.SegmentId.isDataSegmentId;
import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCStatus.CLEANUP;
import static org.apache.jackrabbit.oak.segment.file.PrintableBytes.newPrintableBytes;
import java.io.IOException;
-import java.util.Collection;
import java.util.List;
-import java.util.Set;
-import java.util.UUID;
import org.apache.jackrabbit.guava.common.base.Joiner;
import org.apache.jackrabbit.guava.common.base.Predicate;
-import org.apache.jackrabbit.oak.segment.SegmentId;
import org.apache.jackrabbit.oak.segment.file.tar.CleanupContext;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
@@ -52,7 +46,7 @@ class DefaultCleanupStrategy implements CleanupStrategy {
// to clear stale weak references in the SegmentTracker
System.gc();
- TarFiles.CleanupResult cleanupResult = context.getTarFiles().cleanup(newCleanupContext(context, context.getReclaimer()));
+ TarFiles.CleanupResult cleanupResult = context.getTarFiles().cleanup(newCleanupContext(context));
if (cleanupResult.isInterrupted()) {
context.getGCListener().info("cleanup interrupted");
}
@@ -62,13 +56,17 @@ class DefaultCleanupStrategy implements CleanupStrategy {
long finalSize = size(context);
long reclaimedSize = cleanupResult.getReclaimedSize();
context.getFileStoreStats().reclaimed(reclaimedSize);
- context.getGCJournal().persist(
- reclaimedSize,
- finalSize,
- getGcGeneration(context),
- context.getCompactionMonitor().getCompactedNodes(),
- context.getCompactedRootId()
- );
+
+ GCJournal gcJournal = context.getGCJournal();
+ if (gcJournal != null) {
+ gcJournal.persist(
+ reclaimedSize,
+ finalSize,
+ getGcGeneration(context),
+ context.getCompactionMonitor().getCompactedNodes(),
+ context.getCompactedRootId()
+ );
+ }
context.getGCListener().cleaned(reclaimedSize, finalSize);
context.getGCListener().info(
"cleanup completed in {}. Post cleanup size is {} and space reclaimed {}.",
@@ -79,40 +77,9 @@ class DefaultCleanupStrategy implements CleanupStrategy {
return cleanupResult.getRemovableFiles();
}
- private static CleanupContext newCleanupContext(Context context, Predicate<GCGeneration> old) {
- return new CleanupContext() {
-
- private boolean isUnreferencedBulkSegment(UUID id, boolean referenced) {
- return !isDataSegmentId(id.getLeastSignificantBits()) && !referenced;
- }
-
- private boolean isOldDataSegment(UUID id, GCGeneration generation) {
- return isDataSegmentId(id.getLeastSignificantBits()) && old.apply(generation);
- }
-
- @Override
- public Collection<UUID> initialReferences() {
- Set<UUID> references = newHashSet();
- for (SegmentId id : context.getSegmentTracker().getReferencedSegmentIds()) {
- if (id.isBulkSegmentId()) {
- references.add(id.asUUID());
- }
- }
- return references;
- }
-
- @Override
- public boolean shouldReclaim(UUID id, GCGeneration generation, boolean referenced) {
- return isUnreferencedBulkSegment(id, referenced) || isOldDataSegment(id, generation)
- || generation.getFullGeneration() > getGcGeneration(context).getFullGeneration();
- }
-
- @Override
- public boolean shouldFollow(UUID from, UUID to) {
- return !isDataSegmentId(to.getLeastSignificantBits());
- }
-
- };
+ private static CleanupContext newCleanupContext(Context context) {
+ return new DefaultCleanupContext(context.getSegmentTracker(), context.getReclaimer(),
+ context.getCompactedRootId());
}
private static String toFileNames(@NotNull List<String> files) {
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategy.java
index 342317f0bc..237dbce2fa 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategy.java
@@ -33,7 +33,12 @@ class FullCompactionStrategy extends AbstractCompactionStrategy {
}
@Override
- GCGeneration nextGeneration(GCGeneration current) {
+ GCGeneration partialGeneration(GCGeneration current) {
+ return current.nextPartial();
+ }
+
+ @Override
+ GCGeneration targetGeneration(GCGeneration current) {
return current.nextFull();
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCIncrement.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCIncrement.java
new file mode 100644
index 0000000000..8f925ad575
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCIncrement.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Utility class to keep track of generations for incremental compaction.
+ */
+public class GCIncrement {
+ private final @NotNull GCGeneration baseGeneration;
+ private final @NotNull GCGeneration partialGeneration;
+ private final @NotNull GCGeneration targetGeneration;
+
+ public GCIncrement(@NotNull GCGeneration base, @NotNull GCGeneration partial, @NotNull GCGeneration target) {
+ baseGeneration = base;
+ partialGeneration = partial;
+ targetGeneration = target;
+ }
+
+ @NotNull SegmentWriter createPartialWriter(@NotNull SegmentWriterFactory factory) {
+ return factory.newSegmentWriter(partialGeneration);
+ }
+
+ @NotNull SegmentWriter createTargetWriter(@NotNull SegmentWriterFactory factory) {
+ return factory.newSegmentWriter(targetGeneration);
+ }
+
+ /**
+ * Compaction may be used to copy a repository to the same generation as before.
+ * Therefore, we only consider a segment as fully compacted if it is distinct from the base generation,
+ * even if it already matches the target generation.
+ */
+ boolean isFullyCompacted(GCGeneration generation) {
+ return (generation.compareWith(baseGeneration) > 0) && generation.equals(targetGeneration);
+ }
+
+ @Override
+ public String toString() {
+ return "GCIncrement{\n" +
+ " base: " + baseGeneration + "\n" +
+ " partial: " + partialGeneration + "\n" +
+ " target: " + targetGeneration + "\n" +
+ "}";
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCNodeWriteMonitor.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCNodeWriteMonitor.java
index 7540e68fbc..7613ddf037 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCNodeWriteMonitor.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GCNodeWriteMonitor.java
@@ -48,17 +48,18 @@ public class GCNodeWriteMonitor {
private long estimated = -1;
/**
- * Number of compacted nodes
+ * Number of compacted nodes. This is queried much more often than other properties,
+ * therefore it is the only one to use {@link AtomicLong} instead of {@link LongAdder}.
*/
private long nodes;
/**
- * Number of compacted properties
+ * Number of compacted properties.
*/
private long properties;
/**
- * Number of compacted binaries
+ * Number of compacted binaries.
*/
private long binaries;
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollectionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollectionStrategy.java
index 84f011ad75..eba6c6e824 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollectionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollectionStrategy.java
@@ -27,6 +27,7 @@ import org.apache.jackrabbit.oak.segment.Revisions;
import org.apache.jackrabbit.oak.segment.SegmentCache;
import org.apache.jackrabbit.oak.segment.SegmentReader;
import org.apache.jackrabbit.oak.segment.SegmentTracker;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollector.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollector.java
index 375f0afb1e..bbf471bedd 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollector.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/GarbageCollector.java
@@ -29,12 +29,14 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
+import org.apache.jackrabbit.guava.common.base.Predicate;
import org.apache.jackrabbit.guava.common.base.Supplier;
import org.apache.jackrabbit.oak.segment.Revisions;
import org.apache.jackrabbit.oak.segment.SegmentCache;
import org.apache.jackrabbit.oak.segment.SegmentReader;
import org.apache.jackrabbit.oak.segment.SegmentTracker;
import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.GarbageCollectionStrategy.SuccessfulGarbageCollectionListener;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Reclaimers.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Reclaimers.java
index 08dfe10b6c..fa04f43483 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Reclaimers.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Reclaimers.java
@@ -35,6 +35,19 @@ class Reclaimers {
// Prevent instantiation.
}
+ /**
+ * Create a reclaimer that will never reclaim a segment.
+ */
+ static Predicate<GCGeneration> newEmptyReclaimer() {
+ return new Predicate<GCGeneration>() {
+
+ @Override
+ public boolean apply(GCGeneration generation) {
+ return false;
+ }
+ };
+ }
+
/**
* Create a reclaimer for segments of old generations. Whether a segment is considered old and
* thus reclaimable depends on the type of the most recent GC operation and the number of
@@ -137,7 +150,7 @@ class Reclaimers {
/**
* Create an exact reclaimer. An exact reclaimer reclaims only segment of on single generation.
* @param referenceGeneration the generation to collect.
- * @return an new instance of an exact reclaimer for segments with their generation
+ * @return a new instance of an exact reclaimer for segments with their generation
* matching {@code referenceGeneration}.
*/
static Predicate<GCGeneration> newExactReclaimer(@NotNull final GCGeneration referenceGeneration) {
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TailCompactionStrategy.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TailCompactionStrategy.java
index 8dce6273e9..e4740c4356 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TailCompactionStrategy.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TailCompactionStrategy.java
@@ -35,7 +35,12 @@ class TailCompactionStrategy extends AbstractCompactionStrategy {
}
@Override
- GCGeneration nextGeneration(GCGeneration current) {
+ GCGeneration partialGeneration(GCGeneration current) {
+ return current.nextPartial();
+ }
+
+ @Override
+ GCGeneration targetGeneration(GCGeneration current) {
return current.nextTail();
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/Canceller.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/Canceller.java
index dcd774e232..e395149dfa 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/Canceller.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/Canceller.java
@@ -44,7 +44,15 @@ public class Canceller {
}
Canceller() {
- // Prevent instantiation outside of this package.
+ // Prevent instantiation outside this package.
+ }
+
+ /**
+ * Check if this instance can ever return anything other than {@code NOPE}.
+ * This will be only be false for {@code ROOT} and short circuits thereof.
+ */
+ public boolean isCancelable() {
+ return (this != ROOT);
}
/**
@@ -91,10 +99,9 @@ public class Canceller {
* returned instance will be canceled when this instance is canceled, but
* will never transition back to an "uncanceled" state.
*
- * @return an new instance of {@link Canceller}.
+ * @return a new instance of {@link Canceller}.
*/
public Canceller withShortCircuit() {
return new ShortCircuitCanceller(this);
}
-
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/ShortCircuitCanceller.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/ShortCircuitCanceller.java
index b486ba7cb8..ce4df5ebbb 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/ShortCircuitCanceller.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/cancel/ShortCircuitCanceller.java
@@ -31,9 +31,13 @@ class ShortCircuitCanceller extends Canceller {
this.parent = parent;
}
+ @Override
+ public boolean isCancelable() {
+ return parent.isCancelable();
+ }
+
@Override
public Cancellation check() {
return cancellation.updateAndGet(prev -> prev != null && prev.isCancelled() ? prev : parent.check());
}
-
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/GCGeneration.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/GCGeneration.java
index bb3ad9006b..40e3008a61 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/GCGeneration.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/GCGeneration.java
@@ -21,7 +21,6 @@ package org.apache.jackrabbit.oak.segment.file.tar;
import static org.apache.jackrabbit.guava.common.base.Preconditions.checkNotNull;
import org.apache.jackrabbit.guava.common.base.Objects;
-import org.apache.jackrabbit.oak.segment.file.tar.index.IndexEntry;
import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
import org.jetbrains.annotations.NotNull;
@@ -89,7 +88,7 @@ public final class GCGeneration {
/**
* Create a new instance with the generation and the full generation incremented by one
- * and the compaction flag left unchanged.
+ * and the compaction flag set.
*/
@NotNull
public GCGeneration nextFull() {
@@ -97,14 +96,23 @@ public final class GCGeneration {
}
/**
- * Create a new instance with the generation incremented by one and the full
- * generation and the compaction flag left unchanged.
+ * Create a new instance with the generation incremented by one, the full
+ * generation left unchanged and the compaction flag set.
*/
@NotNull
public GCGeneration nextTail() {
return new GCGeneration(generation + 1, fullGeneration, true);
}
+ /**
+ * Create a new instance with the compaction flag set and the generation and the
+ * full generation left unchanged.
+ */
+ @NotNull
+ public GCGeneration nextPartial() {
+ return new GCGeneration(generation, fullGeneration, true);
+ }
+
/**
* Create a new instance with the compaction flag unset and the generation and the
* full generation left unchanged.
@@ -155,8 +163,8 @@ public final class GCGeneration {
@Override
public String toString() {
return "GCGeneration{" +
- "generation=" + generation + ',' +
- "fullGeneration=" + fullGeneration + ',' +
+ "generation=" + generation + ", " +
+ "fullGeneration=" + fullGeneration + ", " +
"isCompacted=" + isCompacted + '}';
}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/TarReader.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/TarReader.java
index 436ffcd2d1..47f080370d 100644
--- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/TarReader.java
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/tar/TarReader.java
@@ -34,6 +34,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Objects;
import java.util.Set;
import java.util.SortedMap;
import java.util.UUID;
@@ -244,7 +245,7 @@ public class TarReader implements Closeable {
SegmentArchiveReader reader = openStrategy.open(archiveManager, name);
if (reader != null) {
for (String other : archives) {
- if (other != name) {
+ if (!Objects.equals(other, name)) {
log.info("Removing unused tar file {}", other);
archiveManager.delete(other);
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorExternalBlobTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorExternalBlobTest.java
index 8ae3f8453d..a6aac0da21 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorExternalBlobTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorExternalBlobTest.java
@@ -20,6 +20,8 @@ package org.apache.jackrabbit.oak.segment;
import static java.util.concurrent.TimeUnit.DAYS;
import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactor;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameRecord;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameStableId;
@@ -49,21 +51,28 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.RuleChain;
import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.io.File;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+@RunWith(Parameterized.class)
public abstract class AbstractCompactorExternalBlobTest {
- private TemporaryFolder folder = new TemporaryFolder(new File("target"));
+ final private TemporaryFolder folder = new TemporaryFolder(new File("target"));
- private TemporaryBlobStore temporaryBlobStore = new TemporaryBlobStore(folder);
+ final private TemporaryBlobStore temporaryBlobStore = new TemporaryBlobStore(folder);
private FileStore fileStore;
private SegmentNodeStore nodeStore;
- private Compactor compactor;
+ private SimpleCompactor simpleCompactor;
+
+ private final SimpleCompactorFactory compactorFactory;
private GCGeneration compactedGeneration;
@@ -71,6 +80,19 @@ public abstract class AbstractCompactorExternalBlobTest {
public RuleChain rules = RuleChain.outerRule(folder)
.around(temporaryBlobStore);
+ @Parameterized.Parameters
+ public static List<SimpleCompactorFactory> compactorFactories() {
+ return Arrays.asList(
+ compactor -> compactor::compactUp,
+ compactor -> (node, canceller) -> compactor.compactDown(node, canceller, canceller),
+ compactor -> (node, canceller) -> compactor.compact(EMPTY_NODE, node, EMPTY_NODE, canceller)
+ );
+ }
+
+ public AbstractCompactorExternalBlobTest(@NotNull SimpleCompactorFactory compactorFactory) {
+ this.compactorFactory = compactorFactory;
+ }
+
public void setup(boolean withBlobStore) throws IOException, InvalidFileStoreVersionException {
BlobStore blobStore = temporaryBlobStore.blobStore();
FileStoreBuilder fileStoreBuilder = fileStoreBuilder(folder.getRoot());
@@ -82,7 +104,7 @@ public abstract class AbstractCompactorExternalBlobTest {
fileStore = fileStoreBuilder.build();
nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build();
compactedGeneration = newGCGeneration(1,1, true);
- compactor = createCompactor(fileStore, compactedGeneration);
+ simpleCompactor = compactorFactory.newSimpleCompactor(createCompactor(fileStore, compactedGeneration));
}
protected abstract Compactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation);
@@ -117,7 +139,7 @@ public abstract class AbstractCompactorExternalBlobTest {
String cp5 = nodeStore.checkpoint(DAYS.toMillis(1));
SegmentNodeState uncompacted1 = fileStore.getHead();
- SegmentNodeState compacted1 = compactor.compact(EMPTY_NODE, uncompacted1, EMPTY_NODE, Canceller.newCanceller());
+ SegmentNodeState compacted1 = simpleCompactor.compact(uncompacted1, Canceller.newCanceller());
assertNotNull(compacted1);
assertNotSame(uncompacted1, compacted1);
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java
index 77589480ef..298318e362 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java
@@ -20,30 +20,45 @@ package org.apache.jackrabbit.oak.segment;
import static java.util.concurrent.TimeUnit.DAYS;
import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactor;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameRecord;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameStableId;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.checkGeneration;
import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.getCheckpoint;
import static org.apache.jackrabbit.oak.segment.file.FileStoreBuilder.fileStoreBuilder;
-import static org.apache.jackrabbit.oak.segment.file.tar.GCGeneration.newGCGeneration;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.GCIncrement;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.jetbrains.annotations.NotNull;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
public abstract class AbstractCompactorTest {
+
@Rule
public TemporaryFolder folder = new TemporaryFolder(new File("target"));
@@ -53,17 +68,48 @@ public abstract class AbstractCompactorTest {
private Compactor compactor;
- private GCGeneration compactedGeneration;
+ private SimpleCompactor simpleCompactor;
+
+ private final SimpleCompactorFactory compactorFactory;
+
+ private GCNodeWriteMonitor compactionMonitor;
+
+ private GCGeneration baseGeneration;
+ private GCGeneration partialGeneration;
+ private GCGeneration targetGeneration;
+
+ @Parameterized.Parameters
+ public static List<SimpleCompactorFactory> compactorFactories() {
+ return Arrays.asList(
+ compactor -> compactor::compactUp,
+ compactor -> (node, canceller) -> compactor.compactDown(node, canceller, canceller),
+ compactor -> (node, canceller) -> compactor.compact(EMPTY_NODE, node, EMPTY_NODE, canceller)
+ );
+ }
+
+ public AbstractCompactorTest(@NotNull SimpleCompactorFactory compactorFactory) {
+ this.compactorFactory = compactorFactory;
+ }
@Before
public void setup() throws IOException, InvalidFileStoreVersionException {
fileStore = fileStoreBuilder(folder.getRoot()).build();
nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build();
- compactedGeneration = newGCGeneration(1,1, true);
- compactor = createCompactor(fileStore, compactedGeneration);
+
+ baseGeneration = fileStore.getHead().getGcGeneration();
+ partialGeneration = baseGeneration.nextPartial();
+ targetGeneration = baseGeneration.nextFull();
+ GCIncrement increment = new GCIncrement(baseGeneration, partialGeneration, targetGeneration);
+
+ compactionMonitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ compactor = createCompactor(fileStore, increment, compactionMonitor);
+ simpleCompactor = compactorFactory.newSimpleCompactor(compactor);
}
- protected abstract Compactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation);
+ protected abstract Compactor createCompactor(
+ @NotNull FileStore fileStore,
+ @NotNull GCIncrement increment,
+ @NotNull GCNodeWriteMonitor compactionMonitor);
@After
public void tearDown() {
@@ -78,10 +124,10 @@ public abstract class AbstractCompactorTest {
String cp2 = nodeStore.checkpoint(DAYS.toMillis(1));
SegmentNodeState uncompacted1 = fileStore.getHead();
- SegmentNodeState compacted1 = compactor.compact(EMPTY_NODE, uncompacted1, EMPTY_NODE, Canceller.newCanceller());
+ SegmentNodeState compacted1 = simpleCompactor.compact(uncompacted1, Canceller.newCanceller());
assertNotNull(compacted1);
assertNotSame(uncompacted1, compacted1);
- checkGeneration(compacted1, compactedGeneration);
+ checkGeneration(compacted1, targetGeneration);
assertSameStableId(uncompacted1, compacted1);
assertSameStableId(getCheckpoint(uncompacted1, cp1), getCheckpoint(compacted1, cp1));
@@ -98,7 +144,7 @@ public abstract class AbstractCompactorTest {
SegmentNodeState compacted2 = compactor.compact(uncompacted1, uncompacted2, compacted1, Canceller.newCanceller());
assertNotNull(compacted2);
assertNotSame(uncompacted2, compacted2);
- checkGeneration(compacted2, compactedGeneration);
+ checkGeneration(compacted2, targetGeneration);
assertTrue(fileStore.getRevisions().setHead(uncompacted2.getRecordId(), compacted2.getRecordId()));
@@ -112,4 +158,46 @@ public abstract class AbstractCompactorTest {
assertSameRecord(getCheckpoint(compacted1, cp2), getCheckpoint(compacted2, cp2));
assertSameRecord(getCheckpoint(compacted2, cp4), compacted2.getChildNode("root"));
}
+
+ @Test
+ public void testHardCancellation() throws Exception {
+ for (int i = 1; i < 25; i++) {
+ addTestContent("cp" + i, nodeStore, 42);
+ }
+
+ Canceller canceller = Canceller.newCanceller()
+ .withCondition(null, () -> (compactionMonitor.getCompactedNodes() >= 10));
+
+ SegmentNodeState uncompacted1 = fileStore.getHead();
+ SegmentNodeState compacted1 = simpleCompactor.compact(uncompacted1, canceller);
+ assertNull(compacted1);
+ }
+
+ @Test
+ public void testSoftCancellation() throws Exception {
+ for (int i = 1; i < 25; i++) {
+ addTestContent("cp" + i, nodeStore, 42);
+ }
+
+ Canceller nullCanceller = Canceller.newCanceller();
+ Canceller softCanceller = Canceller.newCanceller()
+ .withCondition(null, () -> (compactionMonitor.getCompactedNodes() >= 10));
+
+ SegmentNodeState uncompacted1 = fileStore.getHead();
+ CompactedNodeState compacted1 = compactor.compactDown(uncompacted1, nullCanceller, softCanceller);
+ System.out.println(compactionMonitor.getCompactedNodes());
+
+ assertNotNull(compacted1);
+ assertFalse(compacted1.isComplete());
+ assertEquals(uncompacted1, compacted1);
+ checkGeneration(uncompacted1, baseGeneration);
+ assertNotEquals(targetGeneration, compacted1.getGcGeneration());
+
+ CompactedNodeState compacted2 = simpleCompactor.compact(compacted1, nullCanceller);
+ System.out.println(compactionMonitor.getCompactedNodes());
+
+ assertNotNull(compacted2);
+ assertTrue(compacted2.isComplete());
+ checkGeneration(compacted2, targetGeneration);
+ }
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorExternalBlobTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorExternalBlobTest.java
index baca4d5f04..0bdfb39a5d 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorExternalBlobTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorExternalBlobTest.java
@@ -20,24 +20,26 @@ package org.apache.jackrabbit.oak.segment;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.jetbrains.annotations.NotNull;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
+@RunWith(Parameterized.class)
public class CheckpointCompactorExternalBlobTest extends AbstractCompactorExternalBlobTest {
+ public CheckpointCompactorExternalBlobTest(@NotNull SimpleCompactorFactory compactorFactory) {
+ super(compactorFactory);
+ }
+
@Override
protected CheckpointCompactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation) {
- SegmentWriter writer = defaultSegmentWriterBuilder("c")
- .withGeneration(generation)
- .build(fileStore);
-
- return new CheckpointCompactor(
- GCMonitor.EMPTY,
- fileStore.getReader(),
- writer,
- fileStore.getBlobStore(),
- GCNodeWriteMonitor.EMPTY);
+ SegmentWriter writer = defaultSegmentWriterBuilder("c").withGeneration(generation).build(fileStore);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(), fileStore.getBlobStore(), generation, writer);
+ return new CheckpointCompactor(GCMonitor.EMPTY, compactionWriter, GCNodeWriteMonitor.EMPTY);
}
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
index 95ea97fd72..4c4150a0ce 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
@@ -19,25 +19,33 @@
package org.apache.jackrabbit.oak.segment;
import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.GCIncrement;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
-import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.jetbrains.annotations.NotNull;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
+@RunWith(Parameterized.class)
public class CheckpointCompactorTest extends AbstractCompactorTest {
+ public CheckpointCompactorTest(@NotNull SimpleCompactorFactory compactorFactory) {
+ super(compactorFactory);
+ }
+
@Override
- protected CheckpointCompactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation) {
- SegmentWriter writer = defaultSegmentWriterBuilder("c")
+ protected CheckpointCompactor createCompactor(
+ @NotNull FileStore fileStore,
+ @NotNull GCIncrement increment,
+ @NotNull GCNodeWriteMonitor compactionMonitor
+ ) {
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
.withGeneration(generation)
.build(fileStore);
-
- return new CheckpointCompactor(
- GCMonitor.EMPTY,
- fileStore.getReader(),
- writer,
- fileStore.getBlobStore(),
- GCNodeWriteMonitor.EMPTY);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(), fileStore.getBlobStore(), increment, writerFactory);
+ return new CheckpointCompactor(GCMonitor.EMPTY, compactionWriter, compactionMonitor);
}
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ClassicCompactorTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ClassicCompactorTest.java
index d728c77e18..fc922aff4c 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ClassicCompactorTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ClassicCompactorTest.java
@@ -40,8 +40,10 @@ import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.commons.Buffer;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
@@ -80,7 +82,7 @@ public class ClassicCompactorTest {
addTestContent(nodeStore);
SegmentNodeState uncompacted = (SegmentNodeState) nodeStore.getRoot();
- SegmentNodeState compacted = compactor.compact(uncompacted, Canceller.newCanceller());
+ SegmentNodeState compacted = compactor.compactUp(uncompacted, Canceller.newCanceller());
assertNotNull(compacted);
assertFalse(uncompacted == compacted);
assertEquals(uncompacted, compacted);
@@ -101,7 +103,7 @@ public class ClassicCompactorTest {
addNodes(nodeStore, ClassicCompactor.UPDATE_LIMIT * 2 + 1);
SegmentNodeState uncompacted = (SegmentNodeState) nodeStore.getRoot();
- SegmentNodeState compacted = compactor.compact(uncompacted, Canceller.newCanceller());
+ SegmentNodeState compacted = compactor.compactUp(uncompacted, Canceller.newCanceller());
assertNotNull(compacted);
assertFalse(uncompacted == compacted);
assertEquals(uncompacted, compacted);
@@ -116,25 +118,27 @@ public class ClassicCompactorTest {
builder.setChildNode("cancel").setProperty("cancel", "cancel");
nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
- assertNull(compactor.compact(nodeStore.getRoot(), Canceller.newCanceller().withCondition("reason", () -> true)));
+ assertNull(compactor.compactUp(nodeStore.getRoot(), Canceller.newCanceller().withCondition("reason", () -> true)));
}
@Test(expected = IOException.class)
public void testIOException() throws IOException, CommitFailedException {
ClassicCompactor compactor = createCompactor(fileStore, "IOException");
addTestContent(nodeStore);
- compactor.compact(nodeStore.getRoot(), Canceller.newCanceller());
+ compactor.compactUp(nodeStore.getRoot(), Canceller.newCanceller());
}
@NotNull
private static ClassicCompactor createCompactor(FileStore fileStore, String failOnName) {
+ GCGeneration generation = newGCGeneration(1, 1, true);
SegmentWriter writer = defaultSegmentWriterBuilder("c")
- .withGeneration(newGCGeneration(1, 1, true))
+ .withGeneration(generation)
.build(fileStore);
if (failOnName != null) {
writer = new FailingSegmentWriter(writer, failOnName);
}
- return new ClassicCompactor(fileStore.getReader(), writer, fileStore.getBlobStore(), GCNodeWriteMonitor.EMPTY);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(), fileStore.getBlobStore(), generation, writer);
+ return new ClassicCompactor(compactionWriter, GCNodeWriteMonitor.EMPTY);
}
private static void addNodes(SegmentNodeStore nodeStore, int count)
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CompactorTestUtils.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CompactorTestUtils.java
index 70d06135d4..89aa792477 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CompactorTestUtils.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CompactorTestUtils.java
@@ -20,18 +20,16 @@ package org.apache.jackrabbit.oak.segment;
import static org.apache.jackrabbit.guava.common.collect.Lists.newArrayList;
import static org.apache.jackrabbit.oak.plugins.memory.MultiBinaryPropertyState.binaryPropertyFromBlob;
-import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.CommitFailedException;
-import org.apache.jackrabbit.oak.segment.file.FileStore;
-import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
-import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
@@ -47,6 +45,14 @@ public class CompactorTestUtils {
private CompactorTestUtils() {}
+ public interface SimpleCompactor {
+ CompactedNodeState compact(NodeState nodeState, Canceller canceller) throws IOException;
+ }
+
+ public interface SimpleCompactorFactory {
+ SimpleCompactor newSimpleCompactor(Compactor compactor);
+ }
+
public static void checkGeneration(NodeState node, GCGeneration gcGeneration) {
assertTrue(node instanceof SegmentNodeState);
assertEquals(gcGeneration, ((SegmentNodeState) node).getRecordId().getSegmentId().getGcGeneration());
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
index ae8b817525..210dd60c61 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
@@ -20,16 +20,18 @@ package org.apache.jackrabbit.oak.segment;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.jetbrains.annotations.NotNull;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.List;
import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
@RunWith(Parameterized.class)
public class ParallelCompactorExternalBlobTest extends AbstractCompactorExternalBlobTest {
@@ -37,11 +39,20 @@ public class ParallelCompactorExternalBlobTest extends AbstractCompactorExternal
private final int concurrency;
@Parameterized.Parameters
- public static List<Integer> concurrencyLevels() {
- return Arrays.asList(1, 2, 4, 8, 16);
+ public static List<Object[]> parameters() {
+ Integer[] concurrencyLevels = {1, 2, 4, 8, 16};
+
+ List<Object[]> parameters = new ArrayList<>();
+ for (SimpleCompactorFactory factory : AbstractCompactorExternalBlobTest.compactorFactories()) {
+ for (int concurrency : concurrencyLevels) {
+ parameters.add(new Object[]{factory, concurrency});
+ }
+ }
+ return parameters;
}
- public ParallelCompactorExternalBlobTest(int concurrency) {
+ public ParallelCompactorExternalBlobTest(@NotNull SimpleCompactorFactory compactorFactory, int concurrency) {
+ super(compactorFactory);
this.concurrency = concurrency;
}
@@ -51,13 +62,7 @@ public class ParallelCompactorExternalBlobTest extends AbstractCompactorExternal
.withGeneration(generation)
.withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
.build(fileStore);
-
- return new ParallelCompactor(
- GCMonitor.EMPTY,
- fileStore.getReader(),
- writer,
- fileStore.getBlobStore(),
- GCNodeWriteMonitor.EMPTY,
- concurrency);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(), fileStore.getBlobStore(), generation, writer);
+ return new ParallelCompactor(GCMonitor.EMPTY, compactionWriter, GCNodeWriteMonitor.EMPTY, concurrency);
}
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
index 27bebb1441..2855516ab7 100644
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
@@ -19,17 +19,19 @@
package org.apache.jackrabbit.oak.segment;
import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.GCIncrement;
import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
-import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.jetbrains.annotations.NotNull;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.List;
import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.SimpleCompactorFactory;
@RunWith(Parameterized.class)
public class ParallelCompactorTest extends AbstractCompactorTest {
@@ -37,27 +39,34 @@ public class ParallelCompactorTest extends AbstractCompactorTest {
private final int concurrency;
@Parameterized.Parameters
- public static List<Integer> concurrencyLevels() {
- return Arrays.asList(1, 2, 4, 8, 16);
+ public static List<Object[]> parameters() {
+ Integer[] concurrencyLevels = {1, 2, 4, 8, 16};
+
+ List<Object[]> parameters = new ArrayList<>();
+ for (SimpleCompactorFactory factory : AbstractCompactorExternalBlobTest.compactorFactories()) {
+ for (int concurrency : concurrencyLevels) {
+ parameters.add(new Object[]{factory, concurrency});
+ }
+ }
+ return parameters;
}
- public ParallelCompactorTest(int concurrency) {
+ public ParallelCompactorTest(@NotNull SimpleCompactorFactory compactorFactory, int concurrency) {
+ super(compactorFactory);
this.concurrency = concurrency;
}
@Override
- protected ParallelCompactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation) {
- SegmentWriter writer = defaultSegmentWriterBuilder("c")
+ protected ParallelCompactor createCompactor(
+ @NotNull FileStore fileStore,
+ @NotNull GCIncrement increment,
+ @NotNull GCNodeWriteMonitor compactionMonitor
+ ) {
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
.withGeneration(generation)
.withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
.build(fileStore);
-
- return new ParallelCompactor(
- GCMonitor.EMPTY,
- fileStore.getReader(),
- writer,
- fileStore.getBlobStore(),
- GCNodeWriteMonitor.EMPTY,
- concurrency);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(), fileStore.getBlobStore(), increment, writerFactory);
+ return new ParallelCompactor(GCMonitor.EMPTY, compactionWriter, compactionMonitor, concurrency);
}
}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategyTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategyTest.java
deleted file mode 100644
index c7f67e6338..0000000000
--- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/CompactionStrategyTest.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.jackrabbit.oak.segment.file;
-
-import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-import java.io.IOException;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.when;
-
-public class CompactionStrategyTest {
-
- private static final Throwable MARKER_THROWABLE =
- new RuntimeException("We pretend that something went horribly wrong.");
-
- @Test
- public void compactionIsAbortedOnAnyThrowable() throws IOException {
- MemoryStore store = new MemoryStore();
- CompactionStrategy.Context throwingContext = Mockito.mock(CompactionStrategy.Context.class);
- when(throwingContext.getGCListener()).thenReturn(Mockito.mock(GCListener.class));
- when(throwingContext.getRevisions()).thenReturn(store.getRevisions());
- when(throwingContext.getGCOptions()).thenThrow(MARKER_THROWABLE);
-
- try {
- final CompactionResult compactionResult = new FullCompactionStrategy().compact(throwingContext);
- assertThat("Compaction should be properly aborted.", compactionResult.isSuccess(), is(false));
- } catch (Throwable e) {
- if (e == MARKER_THROWABLE) {
- fail("The marker throwable was not caught by the CompactionStrategy and therefore not properly aborted.");
- } else {
- throw new IllegalStateException("The test likely needs to be adjusted.", e);
- }
- }
- }
-}
\ No newline at end of file
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/DefaultGarbageCollectionStrategyTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/DefaultGarbageCollectionStrategyTest.java
new file mode 100644
index 0000000000..9dc8508236
--- /dev/null
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/DefaultGarbageCollectionStrategyTest.java
@@ -0,0 +1,113 @@
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.SegmentCache;
+import org.apache.jackrabbit.oak.segment.SegmentId;
+import org.apache.jackrabbit.oak.segment.SegmentTracker;
+import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
+import org.apache.jackrabbit.oak.segment.file.tar.CleanupContext;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
+import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.verification.VerificationMode;
+
+import java.io.IOException;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class DefaultGarbageCollectionStrategyTest {
+ private final GCJournal journal;
+
+ public DefaultGarbageCollectionStrategyTest() {
+ journal = Mockito.mock(GCJournal.class);
+ when(journal.read()).thenReturn(Mockito.mock(GCJournal.GCJournalEntry.class));
+ }
+
+ private GarbageCollectionStrategy.Context getMockedGCContext(MemoryStore store) throws IOException {
+ GarbageCollectionStrategy.Context mockedContext = Mockito.mock(GarbageCollectionStrategy.Context.class);
+
+ when(mockedContext.getGCListener()).thenReturn(Mockito.mock(GCListener.class));
+ when(mockedContext.getTarFiles()).thenReturn(Mockito.mock(TarFiles.class));
+ when(mockedContext.getSegmentCache()).thenReturn(Mockito.mock(SegmentCache.class));
+ when(mockedContext.getFileStoreStats()).thenReturn(Mockito.mock(FileStoreStats.class));
+
+ SegmentTracker tracker = new SegmentTracker((msb, lsb) -> new SegmentId(store, msb, lsb));
+ when(mockedContext.getSegmentTracker()).thenReturn(tracker);
+ when(mockedContext.getCompactionMonitor()).thenReturn(GCNodeWriteMonitor.EMPTY);
+ when(mockedContext.getRevisions()).thenReturn(store.getRevisions());
+ when(mockedContext.getGCJournal()).thenReturn(journal);
+
+ TarFiles mockedTarFiles = Mockito.mock(TarFiles.class);
+ when(mockedContext.getTarFiles()).thenReturn(mockedTarFiles);
+ when(mockedTarFiles.cleanup(any(CleanupContext.class)))
+ .thenReturn(Mockito.mock(TarFiles.CleanupResult.class));
+
+ return mockedContext;
+ }
+
+ private void runCleanup(CompactionResult result) throws IOException {
+ MemoryStore store = new MemoryStore();
+ DefaultGarbageCollectionStrategy strategy = new DefaultGarbageCollectionStrategy();
+ strategy.cleanup(getMockedGCContext(store), result);
+ }
+
+ private void verifyGCJournalPersistence(VerificationMode mode) {
+ verify(journal, mode).persist(
+ anyLong(),
+ anyLong(),
+ any(GCGeneration.class),
+ anyLong(),
+ anyString());
+ }
+
+ @Test
+ public void successfulCompactionPersistsToJournal() throws Exception {
+ CompactionResult result = CompactionResult.succeeded(
+ SegmentGCOptions.GCType.FULL,
+ GCGeneration.NULL,
+ SegmentGCOptions.defaultGCOptions(),
+ RecordId.NULL,
+ 0);
+ runCleanup(result);
+ verifyGCJournalPersistence(times(1));
+ }
+
+ @Test
+ public void partialCompactionDoesNotPersistToJournal() throws Exception {
+ CompactionResult result = CompactionResult.partiallySucceeded(GCGeneration.NULL, RecordId.NULL, 0);
+ runCleanup(result);
+ verifyGCJournalPersistence(never());
+ }
+
+ @Test
+ public void skippedCompactionDoesNotPersistToJournal() throws Exception {
+ CompactionResult result = CompactionResult.skipped(
+ SegmentGCOptions.GCType.FULL,
+ GCGeneration.NULL,
+ SegmentGCOptions.defaultGCOptions(),
+ RecordId.NULL,
+ 0);
+ runCleanup(result);
+ verifyGCJournalPersistence(never());
+ }
+
+ @Test
+ public void nonApplicableCompactionDoesNotPersistToJournal() throws Exception {
+ runCleanup(CompactionResult.notApplicable(0));
+ verifyGCJournalPersistence(never());
+ }
+
+ @Test
+ public void abortedCompactionDoesNotPersistToJournal() throws Exception {
+ runCleanup(CompactionResult.aborted(GCGeneration.NULL, 0));
+ verifyGCJournalPersistence(never());
+ }
+}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategyTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategyTest.java
new file mode 100644
index 0000000000..44995f6846
--- /dev/null
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/FullCompactionStrategyTest.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.CompactorTestUtils;
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.SegmentBufferWriterPool;
+import org.apache.jackrabbit.oak.segment.SegmentNodeState;
+import org.apache.jackrabbit.oak.segment.SegmentNodeStoreBuilders;
+import org.apache.jackrabbit.oak.segment.SegmentWriterFactory;
+import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
+import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
+import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+public class FullCompactionStrategyTest {
+
+ private static final Throwable MARKER_THROWABLE =
+ new RuntimeException("We pretend that something went horribly wrong.");
+
+ @Test
+ public void compactionIsAbortedOnAnyThrowable() throws IOException {
+ MemoryStore store = new MemoryStore();
+ CompactionStrategy.Context throwingContext = Mockito.mock(CompactionStrategy.Context.class);
+ when(throwingContext.getGCListener()).thenReturn(Mockito.mock(GCListener.class));
+ when(throwingContext.getRevisions()).thenReturn(store.getRevisions());
+ when(throwingContext.getGCOptions()).thenThrow(MARKER_THROWABLE);
+
+ try {
+ final CompactionResult compactionResult = new FullCompactionStrategy().compact(throwingContext);
+ assertThat("Compaction should be properly aborted.", compactionResult.isSuccess(), is(false));
+ } catch (Throwable e) {
+ if (e == MARKER_THROWABLE) {
+ fail("The marker throwable was not caught by the CompactionStrategy and therefore not properly aborted.");
+ } else {
+ throw new IllegalStateException("The test likely needs to be adjusted.", e);
+ }
+ }
+ }
+
+ private CompactionStrategy.Context getMockedCompactionContext(MemoryStore store) {
+ CompactionStrategy.Context mockedContext = Mockito.mock(CompactionStrategy.Context.class);
+ when(mockedContext.getGCListener()).thenReturn(Mockito.mock(GCListener.class));
+ when(mockedContext.getTarFiles()).thenReturn(Mockito.mock(TarFiles.class));
+ when(mockedContext.getSuccessfulCompactionListener()).thenReturn(Mockito.mock(SuccessfulCompactionListener.class));
+ when(mockedContext.getGCOptions()).thenReturn(SegmentGCOptions.defaultGCOptions());
+ when(mockedContext.getFlusher()).thenReturn(() -> {});
+
+ GCJournal mockedJournal = Mockito.mock(GCJournal.class);
+ when(mockedContext.getGCJournal()).thenReturn(mockedJournal);
+ when(mockedJournal.read()).thenReturn(Mockito.mock(GCJournal.GCJournalEntry.class));
+
+ when(mockedContext.getRevisions()).thenReturn(store.getRevisions());
+ when(mockedContext.getSegmentReader()).thenReturn(store.getReader());
+
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
+ .withGeneration(generation)
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
+ .build(store);
+ when(mockedContext.getSegmentWriterFactory()).thenReturn(writerFactory);
+
+ return mockedContext;
+ }
+
+ @Test
+ public void testIntermediateStateSave() throws Exception {
+ MemoryStore store = new MemoryStore();
+ NodeStore nodeStore = SegmentNodeStoreBuilders.builder(store).build();
+
+ for (int i = 1; i < 100; i++) {
+ addTestContent("node" + i, nodeStore, 42);
+ }
+
+ CompactionStrategy.Context mockedContext = getMockedCompactionContext(store);
+ GCNodeWriteMonitor gcMonitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ when(mockedContext.getCompactionMonitor()).thenReturn(gcMonitor);
+
+ when(mockedContext.getStateSaveTriggerSupplier()).thenReturn(
+ () -> {
+ long compactedNodes = gcMonitor.getCompactedNodes();
+ return Canceller.newCanceller().withCondition("10 more nodes compacted",
+ () -> gcMonitor.getCompactedNodes() >= compactedNodes + 10);
+ }
+ );
+ when(mockedContext.getHardCanceller()).thenReturn(
+ Canceller.newCanceller().withCondition("50 total nodes compacted",
+ () -> gcMonitor.getCompactedNodes() >= 50)
+ );
+ when(mockedContext.getSoftCanceller()).thenReturn(Canceller.newCanceller());
+
+ RecordId initialHead = store.getRevisions().getHead();
+
+ FullCompactionStrategy strategy = new FullCompactionStrategy();
+ CompactionResult result = strategy.compact(mockedContext);
+
+ assertFalse(result.isSuccess());
+ assertNotEquals(initialHead, store.getRevisions().getHead());
+ }
+
+ @Test
+ public void testIncrementalCompaction() throws Exception {
+ MemoryStore store = new MemoryStore();
+ NodeStore nodeStore = SegmentNodeStoreBuilders.builder(store).build();
+
+ for (int i = 1; i < 100; i++) {
+ addTestContent("node" + i, nodeStore, 42);
+ }
+
+ CompactionStrategy.Context mockedContext = getMockedCompactionContext(store);
+ GCNodeWriteMonitor gcMonitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ when(mockedContext.getCompactionMonitor()).thenReturn(gcMonitor);
+
+ when(mockedContext.getStateSaveTriggerSupplier()).thenReturn(
+ () -> {
+ long compactedNodes = gcMonitor.getCompactedNodes();
+ return Canceller.newCanceller().withCondition("10 more nodes compacted",
+ () -> gcMonitor.getCompactedNodes() >= compactedNodes + 10);
+ }
+ );
+ when(mockedContext.getHardCanceller()).thenReturn(Canceller.newCanceller());
+ when(mockedContext.getSoftCanceller()).thenReturn(Canceller.newCanceller());
+
+ SegmentNodeState initialHeadState = store.getReader().readNode(store.getRevisions().getHead());
+ GCGeneration baseGeneration = initialHeadState.getGcGeneration();
+ CompactorTestUtils.checkGeneration(initialHeadState, baseGeneration);
+
+ FullCompactionStrategy strategy = new FullCompactionStrategy();
+ CompactionResult result = strategy.compact(mockedContext);
+ assertTrue(result.isSuccess());
+
+ SegmentNodeState headState = store.getReader().readNode(store.getRevisions().getHead());
+ CompactorTestUtils.checkGeneration(headState, baseGeneration.nextFull());
+ }
+}
\ No newline at end of file
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/GCIncrementTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/GCIncrementTest.java
new file mode 100644
index 0000000000..469f7a74c1
--- /dev/null
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/GCIncrementTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.junit.Test;
+
+import static org.apache.jackrabbit.oak.segment.file.tar.GCGeneration.newGCGeneration;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class GCIncrementTest {
+
+ @Test
+ public void testIsCompactedSameGeneration() {
+ GCGeneration base = GCGeneration.NULL;
+ GCIncrement increment = new GCIncrement(base, base, base);
+
+ assertFalse(increment.isFullyCompacted(base));
+ assertFalse(increment.isFullyCompacted(newGCGeneration(1, 1, true)));
+ }
+
+ @Test
+ public void testIsCompactedTail() {
+ GCGeneration base = GCGeneration.NULL;
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), base.nextTail());
+
+ assertFalse(increment.isFullyCompacted(newGCGeneration(1, 0, false)));
+ assertFalse(increment.isFullyCompacted(base));
+ assertFalse(increment.isFullyCompacted(base.nextPartial()));
+ assertTrue(increment.isFullyCompacted(base.nextTail()));
+ }
+
+ @Test
+ public void testIsCompactedFull() {
+ GCGeneration base = GCGeneration.NULL;
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), base.nextFull());
+
+ assertFalse(increment.isFullyCompacted(base));
+ assertFalse(increment.isFullyCompacted(newGCGeneration(1, 1, false)));
+ assertFalse(increment.isFullyCompacted(base.nextPartial()));
+ assertFalse(increment.isFullyCompacted(base.nextTail()));
+ assertTrue(increment.isFullyCompacted(base.nextFull()));
+ }
+
+}