[ASTERIXDB-2169][STO][TX] Unblock modifications during full scan - user model changes: no - storage format changes: no - interface changes: yes - added ILSMHarness.replaceMemoryComponentsWithDiskComponents
details: - During a long running query aka full scan, two things block incoming modifications: 1) Memory component gets full, is flushed but can't be recycled because of the search operation inside the component. 2) Read latches on the memory component not being released and the memory component search cursor is not advancing. The two cases are addressed in this change for the LSMBTree but not yet addressed for other indexes. The proposed solution for case (1) is to poll memory components states every n records during the search operation. If a memory component was found to have been flushed, its cursor is moved to the corresponding disk component allowing the memory component to be recycled. The proposed solution for case (2) is to check memory component cursor every n records. If the cursor has not advanced and the component has writers, then the latches over the leaf page are released, and the cursor re-do the operation entering from the tree root. - Added a test case. - Added performance traces for enter and exit components. Change-Id: I37ba52f6324ed1c5a78465c3a8cbcd351f1ed5bc Reviewed-on: https://asterix-gerrit.ics.uci.edu/2166 Sonar-Qube: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Tested-by: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Contrib: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Integration-Tests: Jenkins <jenk...@fulliautomatix.ics.uci.edu> Reviewed-by: Luo Chen <cl...@uci.edu> Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/e5a65429 Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/e5a65429 Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/e5a65429 Branch: refs/heads/master Commit: e5a65429d94c34b3079a3fe6bad206089c2c193c Parents: 76ecc4b Author: Abdullah Alamoudi <bamou...@gmail.com> Authored: Mon Nov 20 17:14:01 2017 -0800 Committer: abdullah alamoudi <bamou...@gmail.com> Committed: Mon Nov 20 21:32:04 2017 -0800 ---------------------------------------------------------------------- .../app/bootstrap/TestNodeController.java | 58 +-- .../test/dataflow/ComponentRollbackTest.java | 94 +++-- .../asterix/test/dataflow/LogMarkerTest.java | 7 +- .../dataflow/MultiPartitionLSMIndexTest.java | 196 ++++++++++ .../SearchCursorComponentSwitchTest.java | 266 +++++++++++++ .../asterix/test/dataflow/TestDataset.java | 3 +- .../TestLsmBtreeIoOpCallbackFactory.java | 8 +- .../asterix/test/logging/CheckpointingTest.java | 7 +- .../asterix/test/storage/DeallocatableTest.java | 4 +- .../asterix/test/storage/DiskIsFullTest.java | 8 +- .../IndexDropOperatorNodePushableTest.java | 28 +- .../AbstractLSMIOOperationCallback.java | 10 +- ...actOneInputOneOutputOneFramePushRuntime.java | 16 +- .../meta/AlgebricksMetaOperatorDescriptor.java | 49 ++- .../apache/hyracks/http/server/HttpServer.java | 10 +- .../dataflow/ExternalBTreeLocalResource.java | 2 +- .../ExternalBTreeWithBuddyLocalResource.java | 2 +- .../am/lsm/btree/impls/ExternalBTree.java | 27 +- .../lsm/btree/impls/ExternalBTreeOpContext.java | 8 +- .../lsm/btree/impls/ExternalBTreeWithBuddy.java | 26 +- .../impls/ExternalBTreeWithBuddyOpContext.java | 9 +- .../storage/am/lsm/btree/impls/LSMBTree.java | 14 +- .../am/lsm/btree/impls/LSMBTreeOpContext.java | 13 +- .../btree/impls/LSMBTreeRangeSearchCursor.java | 152 +++++++- .../am/lsm/btree/utils/LSMBTreeUtil.java | 9 +- .../am/lsm/common/api/ILSMComponent.java | 5 + .../storage/am/lsm/common/api/ILSMHarness.java | 13 + .../storage/am/lsm/common/api/ILSMIndex.java | 5 + .../common/api/ILSMIndexOperationContext.java | 26 ++ .../lsm/common/impls/AbstractLSMComponent.java | 7 +- .../common/impls/AbstractLSMDiskComponent.java | 9 +- .../am/lsm/common/impls/AbstractLSMIndex.java | 39 +- .../impls/AbstractLSMIndexOperationContext.java | 42 ++- .../impls/AbstractLSMMemoryComponent.java | 6 + .../impls/ComponentReplacementContext.java | 215 +++++++++++ .../am/lsm/common/impls/LSMComponentId.java | 8 +- .../storage/am/lsm/common/impls/LSMHarness.java | 373 +++++++++++-------- .../impls/LSMIndexDiskComponentBulkLoader.java | 2 +- .../common/impls/LSMIndexReplicationJob.java | 2 +- .../lsm/common/impls/LSMIndexSearchCursor.java | 19 +- .../invertedindex/impls/LSMInvertedIndex.java | 14 +- .../impls/LSMInvertedIndexOpContext.java | 11 +- .../ondisk/OnDiskInvertedIndexOpContext.java | 1 + .../dataflow/ExternalRTreeLocalResource.java | 2 +- .../am/lsm/rtree/impls/AbstractLSMRTree.java | 10 +- .../am/lsm/rtree/impls/ExternalRTree.java | 29 +- .../lsm/rtree/impls/ExternalRTreeOpContext.java | 9 +- .../storage/am/lsm/rtree/impls/LSMRTree.java | 11 +- .../am/lsm/rtree/impls/LSMRTreeOpContext.java | 16 +- .../impls/LSMRTreeWithAntiMatterTuples.java | 6 +- .../am/lsm/rtree/utils/LSMRTreeUtils.java | 5 +- .../storage/am/lsm/btree/impl/TestLsmBtree.java | 25 +- 52 files changed, 1514 insertions(+), 422 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java index 90b9a1e..505ef8f 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java @@ -76,6 +76,8 @@ import org.apache.hyracks.api.context.IHyracksJobletContext; import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.dataflow.ActivityId; import org.apache.hyracks.api.dataflow.OperatorDescriptorId; +import org.apache.hyracks.api.dataflow.TaskAttemptId; +import org.apache.hyracks.api.dataflow.TaskId; import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider; import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer; import org.apache.hyracks.api.dataflow.value.ITypeTraits; @@ -121,11 +123,10 @@ public class TestNodeController { public static final int DEFAULT_HYRACKS_CC_CLIENT_PORT = 1098; public static final int DEFAULT_HYRACKS_CC_CLUSTER_PORT = 1099; public static final int KB32 = 32768; - public static final int PARTITION = 0; public static final double BLOOM_FILTER_FALSE_POSITIVE_RATE = 0.01; public static final TransactionSubsystemProvider TXN_SUBSYSTEM_PROVIDER = TransactionSubsystemProvider.INSTANCE; // Mutables - private long jobCounter = 0L; + private long jobCounter = 100L; private final String testConfigFileName; private final boolean runHDFS; @@ -155,7 +156,11 @@ public class TestNodeController { } public TxnId getTxnJobId(IHyracksTaskContext ctx) { - return new TxnId(ctx.getJobletContext().getJobId().getId()); + return getTxnJobId(ctx.getJobletContext().getJobId()); + } + + public TxnId getTxnJobId(JobId jobId) { + return new TxnId(jobId.getId()); } public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx, @@ -177,12 +182,14 @@ public class TestNodeController { IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider(); IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory( storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider()); - LSMInsertDeleteOperatorNodePushable insertOp = new LSMInsertDeleteOperatorNodePushable(ctx, PARTITION, - primaryIndexInfo.primaryIndexInsertFieldsPermutations, - recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), op, - true, indexHelperFactory, modOpCallbackFactory, null); - CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), - primaryIndexInfo.primaryKeyIndexes, false, true, PARTITION, true); + LSMInsertDeleteOperatorNodePushable insertOp = + new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(), + primaryIndexInfo.primaryIndexInsertFieldsPermutations, + recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), + op, true, indexHelperFactory, modOpCallbackFactory, null); + CommitRuntime commitOp = + new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), primaryIndexInfo.primaryKeyIndexes, + false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true); insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc); commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc); return Pair.of(insertOp, commitOp); @@ -203,7 +210,8 @@ public class TestNodeController { null, null, true, true, indexDataflowHelperFactory, false, false, null, NoOpOperationCallbackFactory.INSTANCE, filterFields, filterFields, false); BTreeSearchOperatorNodePushable searchOp = - searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(), PARTITION, 1); + searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(), + ctx.getTaskAttemptId().getTaskId().getPartition(), 1); emptyTupleOp.setOutputFrameWriter(0, searchOp, primaryIndexInfo.getSearchRecordDescriptorProvider().getInputRecordDescriptor(null, 0)); searchOp.setOutputFrameWriter(0, countOp, primaryIndexInfo.rDesc); @@ -236,7 +244,7 @@ public class TestNodeController { public PrimaryIndexInfo createPrimaryIndex(Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType, int[] filterFields, IStorageComponentProvider storageComponentProvider, - int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators) + int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators, int partition) throws AlgebricksException, HyracksDataException, RemoteException, ACIDException { MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction(); org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy = @@ -254,8 +262,8 @@ public class TestNodeController { IndexBuilderFactory indexBuilderFactory = new IndexBuilderFactory(storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider(), resourceFactory, !dataset.isTemp()); - IHyracksTaskContext ctx = createTestContext(false); - IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, 0); + IHyracksTaskContext ctx = createTestContext(newJobId(), partition, false); + IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, partition); indexBuilder.build(); } finally { mdProvider.getLocks().unlock(); @@ -292,12 +300,12 @@ public class TestNodeController { return primaryIndexTypeTraits; } - public IHyracksTaskContext createTestContext(boolean withMessaging) throws HyracksDataException { + public IHyracksTaskContext createTestContext(JobId jobId, int partition, boolean withMessaging) + throws HyracksDataException { IHyracksTaskContext ctx = TestUtils.create(KB32); if (withMessaging) { TaskUtil.put(HyracksConstants.KEY_MESSAGE, new VSizeFrame(ctx), ctx); } - JobId jobId = newJobId(); IHyracksJobletContext jobletCtx = Mockito.mock(IHyracksJobletContext.class); JobEventListenerFactory factory = new JobEventListenerFactory(new TxnId(jobId.getId()), true); Mockito.when(jobletCtx.getJobletEventListenerFactory()).thenReturn(factory); @@ -306,6 +314,9 @@ public class TestNodeController { ctx = Mockito.spy(ctx); Mockito.when(ctx.getJobletContext()).thenReturn(jobletCtx); Mockito.when(ctx.getIoManager()).thenReturn(ExecutionTestUtil.integrationUtil.ncs[0].getIoManager()); + TaskAttemptId taskId = + new TaskAttemptId(new TaskId(new ActivityId(new OperatorDescriptorId(0), 0), partition), 0); + Mockito.when(ctx.getTaskAttemptId()).thenReturn(taskId); return ctx; } @@ -377,7 +388,7 @@ public class TestNodeController { (CcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext(); FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset, index.getIndexName(), nodes); - fileSplitProvider = new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1)); + fileSplitProvider = new ConstantFileSplitProvider(splits); } public Index getIndex() { @@ -448,12 +459,13 @@ public class TestNodeController { IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider(); IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory( storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider()); - LSMPrimaryUpsertOperatorNodePushable insertOp = new LSMPrimaryUpsertOperatorNodePushable(ctx, PARTITION, - indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations, - recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), - modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1, - frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory, - MissingWriterFactory.INSTANCE, hasSecondaries); + LSMPrimaryUpsertOperatorNodePushable insertOp = + new LSMPrimaryUpsertOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(), + indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations, + recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), + modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1, + frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory, + MissingWriterFactory.INSTANCE, hasSecondaries); RecordDescriptor upsertOutRecDesc = getUpsertOutRecDesc(primaryIndexInfo.rDesc, dataset, filterFields == null ? 0 : filterFields.length, recordType, metaType); // fix pk fields @@ -463,7 +475,7 @@ public class TestNodeController { pkFieldsInCommitOp[i] = diff + i; } CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), pkFieldsInCommitOp, - false, true, PARTITION, true); + false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true); insertOp.setOutputFrameWriter(0, commitOp, upsertOutRecDesc); commitOp.setInputRecordDescriptor(0, upsertOutRecDesc); return Pair.of(insertOp, commitOp); http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java index b09e000..1ac1aa6 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java @@ -58,6 +58,7 @@ import org.apache.hyracks.api.comm.VSizeFrame; import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.dataflow.value.RecordDescriptor; import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.api.test.CountAnswer; import org.apache.hyracks.api.test.FrameWriterTestUtils; import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation; @@ -133,20 +134,21 @@ public class ComponentRollbackTest { public void createIndex() throws Exception { List<List<String>> partitioningKeys = new ArrayList<>(); partitioningKeys.add(Collections.singletonList("key")); + int partition = 0; dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME, NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH, partitioningKeys, null, null, null, false, null, false), null, DatasetType.INTERNAL, DATASET_ID, 0); PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, - storageManager, KEY_INDEXES, KEY_INDICATORS_LIST); + storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, partition); IndexDataflowHelperFactory iHelperFactory = new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider()); - ctx = nc.createTestContext(false); - indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0); + JobId jobId = nc.newJobId(); + ctx = nc.createTestContext(jobId, partition, false); + indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), partition); indexDataflowHelper.open(); lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance(); indexDataflowHelper.close(); - nc.newJobId(); txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx), new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL)); insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES, @@ -158,7 +160,7 @@ public class ComponentRollbackTest { indexDataflowHelper.destroy(); } - private void allowAllOps(TestLsmBtree lsmBtree) { + static void allowAllOps(TestLsmBtree lsmBtree) { lsmBtree.addModifyCallback(sem -> sem.release()); lsmBtree.addFlushCallback(sem -> sem.release()); lsmBtree.addSearchCallback(sem -> sem.release()); @@ -197,19 +199,17 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); // rollback a memory component lsmAccessor.deleteComponents(memoryComponentsPredicate); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); // rollback the last disk component lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata()); DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); lsmAccessor.deleteComponents(pred); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -247,12 +247,11 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); // rollback a memory component lsmAccessor.deleteComponents(memoryComponentsPredicate); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); // insert again nc.newJobId(); @@ -270,14 +269,13 @@ public class ComponentRollbackTest { } insertOp.close(); nc.getTransactionManager().commitTransaction(txnCtx.getTxnId()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // rollback the last disk component lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata()); DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); lsmAccessor.deleteComponents(pred); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -317,12 +315,11 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); // wait till firstSearcher enter the components firstSearcher.waitUntilEntered(); // now that we enetered, we will rollback ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); // rollback a memory component lsmAccessor.deleteComponents( c -> (c instanceof ILSMMemoryComponent && ((ILSMMemoryComponent) c).isModified())); @@ -331,24 +328,23 @@ public class ComponentRollbackTest { lsmBtree.allowSearch(1); Assert.assertTrue(firstSearcher.result()); // search now and ensure - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); // rollback the last disk component // re-block searches lsmBtree.clearSearchCallbacks(); - Searcher secondSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, + Searcher secondSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT); // wait till firstSearcher enter the components secondSearcher.waitUntilEntered(); lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata()); DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); lsmAccessor.deleteComponents(pred); // now that the rollback has completed, we will unblock the search lsmBtree.addSearchCallback(sem -> sem.release()); lsmBtree.allowSearch(1); Assert.assertTrue(secondSearcher.result()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT)); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -386,7 +382,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // disable flushes lsmBtree.clearFlushCallbacks(); Flusher firstFlusher = new Flusher(lsmBtree); @@ -395,7 +391,7 @@ public class ComponentRollbackTest { // now that we enetered, we will rollback. This will not proceed since it is waiting for the flush to complete Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate); // now that the rollback has completed, we will search - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); //unblock the flush lsmBtree.allowFlush(1); // ensure rollback completed @@ -403,7 +399,7 @@ public class ComponentRollbackTest { // ensure current mem component is not modified Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); // search now and ensure - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -442,7 +438,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // Now, we will start a full merge Merger merger = new Merger(lsmBtree); ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); @@ -459,7 +455,7 @@ public class ComponentRollbackTest { Rollerback rollerback = new Rollerback(lsmBtree, new DiskComponentLsnPredicate(lsn)); // rollback is now waiting for the merge to complete // we will search - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); //unblock the merge lsmBtree.allowMerge(1); // ensure rollback completes @@ -467,7 +463,7 @@ public class ComponentRollbackTest { // ensure current mem component is not modified Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); // search now and ensure that we rolled back the merged component - searchAndAssertCount(nc, ctx, dataset, storageManager, + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT)); } catch (Throwable e) { e.printStackTrace(); @@ -506,7 +502,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // disable flushes // disable searches lsmBtree.clearFlushCallbacks(); @@ -514,7 +510,7 @@ public class ComponentRollbackTest { Flusher firstFlusher = new Flusher(lsmBtree); dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true); firstFlusher.waitUntilCount(1); - Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); // wait till firstSearcher enter the components firstSearcher.waitUntilEntered(); // now that we enetered, we will rollback rollback a memory component @@ -528,7 +524,7 @@ public class ComponentRollbackTest { rollerback.complete(); Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); // search now and ensure the rollback was no op since it waits for ongoing flushes - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -566,7 +562,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // disable flushes // disable searches lsmBtree.clearFlushCallbacks(); @@ -574,7 +570,7 @@ public class ComponentRollbackTest { dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true); firstFlusher.waitUntilCount(1); lsmBtree.clearSearchCallbacks(); - Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); // wait till firstSearcher enter the components firstSearcher.waitUntilEntered(); // now that we enetered, we will rollback @@ -589,7 +585,7 @@ public class ComponentRollbackTest { rollerback.complete(); Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); // search now and ensure - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); } catch (Throwable e) { e.printStackTrace(); Assert.fail(e.getMessage()); @@ -628,7 +624,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // Now, we will start a merge Merger merger = new Merger(lsmBtree); ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); @@ -643,7 +639,7 @@ public class ComponentRollbackTest { merger.waitUntilCount(1); // we will block search lsmBtree.clearSearchCallbacks(); - Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); // wait till firstSearcher enter the components firstSearcher.waitUntilEntered(); // now that we enetered, we will rollback @@ -657,7 +653,7 @@ public class ComponentRollbackTest { Assert.assertTrue(firstSearcher.result()); rollerback.complete(); // now that the rollback has completed, we will search - searchAndAssertCount(nc, ctx, dataset, storageManager, + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT)); // ensure current mem component is not modified Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); @@ -699,7 +695,7 @@ public class ComponentRollbackTest { List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents(); Assert.assertEquals(9, diskComponents.size()); Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); // Now, we will start a merge Merger merger = new Merger(lsmBtree); ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); @@ -714,7 +710,7 @@ public class ComponentRollbackTest { merger.waitUntilCount(1); // we will block search lsmBtree.clearSearchCallbacks(); - Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); // wait till firstSearcher enter the components firstSearcher.waitUntilEntered(); // now that we enetered, we will rollback @@ -724,11 +720,11 @@ public class ComponentRollbackTest { lsmBtree.allowSearch(1); Assert.assertTrue(firstSearcher.result()); // even though rollback has been called, it is still waiting for the merge to complete - searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); //unblock the merge lsmBtree.allowMerge(1); rollerBack.complete(); - searchAndAssertCount(nc, ctx, dataset, storageManager, + searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT)); // ensure current mem component is not modified Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified()); @@ -748,7 +744,6 @@ public class ComponentRollbackTest { @Override public void run() { ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); - dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh(); try { lsmAccessor.deleteComponents(predicate); } catch (HyracksDataException e) { @@ -768,13 +763,13 @@ public class ComponentRollbackTest { } } - private class Searcher { + static class Searcher { private final ExecutorService executor = Executors.newSingleThreadExecutor(); private Future<Boolean> task; private volatile boolean entered = false; - public Searcher(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset, - StorageComponentProvider storageManager, TestLsmBtree lsmBtree, int numOfRecords) { + public Searcher(TestNodeController nc, int partition, Dataset dataset, StorageComponentProvider storageManager, + TestLsmBtree lsmBtree, int numOfRecords) { lsmBtree.addSearchCallback(sem -> { synchronized (Searcher.this) { entered = true; @@ -784,7 +779,7 @@ public class ComponentRollbackTest { Callable<Boolean> callable = new Callable<Boolean>() { @Override public Boolean call() throws Exception { - searchAndAssertCount(nc, ctx, dataset, storageManager, numOfRecords); + searchAndAssertCount(nc, partition, dataset, storageManager, numOfRecords); return true; } }; @@ -840,7 +835,7 @@ public class ComponentRollbackTest { } } - private class DiskComponentLsnPredicate implements Predicate<ILSMComponent> { + private static class DiskComponentLsnPredicate implements Predicate<ILSMComponent> { private final long lsn; public DiskComponentLsnPredicate(long lsn) { @@ -860,10 +855,11 @@ public class ComponentRollbackTest { } } - private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset, + static void searchAndAssertCount(TestNodeController nc, int partition, Dataset dataset, StorageComponentProvider storageManager, int numOfRecords) throws HyracksDataException, AlgebricksException { - nc.newJobId(); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, partition, false); TestTupleCounterFrameWriter countOp = create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE), Collections.emptyList(), Collections.emptyList(), false); IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java index 1907ed6..aa08c2d 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java @@ -50,6 +50,7 @@ import org.apache.hyracks.algebricks.runtime.base.IPushRuntime; import org.apache.hyracks.api.comm.VSizeFrame; import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.dataflow.value.RecordDescriptor; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.api.test.CountAnswer; import org.apache.hyracks.api.test.FrameWriterTestUtils; import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation; @@ -117,9 +118,9 @@ public class LogMarkerTest { null, DatasetType.INTERNAL, DATASET_ID, 0); try { PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, - storageManager, KEY_INDEXES, KEY_INDICATORS_LIST); - IHyracksTaskContext ctx = nc.createTestContext(true); - nc.newJobId(); + storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true); ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx), new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL)); LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java new file mode 100644 index 0000000..ec9c2f6 --- /dev/null +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java @@ -0,0 +1,196 @@ +/* + * 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.asterix.test.dataflow; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.asterix.app.bootstrap.TestNodeController; +import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo; +import org.apache.asterix.app.data.gen.TupleGenerator; +import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction; +import org.apache.asterix.app.nc.NCAppRuntimeContext; +import org.apache.asterix.common.api.IDatasetLifecycleManager; +import org.apache.asterix.common.config.DatasetConfig.DatasetType; +import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable; +import org.apache.asterix.common.transactions.ITransactionContext; +import org.apache.asterix.common.transactions.ITransactionManager; +import org.apache.asterix.common.transactions.TransactionOptions; +import org.apache.asterix.external.util.DataflowUtils; +import org.apache.asterix.file.StorageComponentProvider; +import org.apache.asterix.metadata.entities.Dataset; +import org.apache.asterix.metadata.entities.Index; +import org.apache.asterix.metadata.entities.InternalDatasetDetails; +import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy; +import org.apache.asterix.om.types.ARecordType; +import org.apache.asterix.om.types.BuiltinType; +import org.apache.asterix.om.types.IAType; +import org.apache.asterix.test.common.TestHelper; +import org.apache.hyracks.api.comm.VSizeFrame; +import org.apache.hyracks.api.context.IHyracksTaskContext; +import org.apache.hyracks.api.job.JobId; +import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender; +import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference; +import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper; +import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory; +import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree; +import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class MultiPartitionLSMIndexTest { + private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 }; + private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" }, + new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false); + private static final GenerationFunction[] RECORD_GEN_FUNCTION = + { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC }; + private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false }; + private static final ARecordType META_TYPE = null; + private static final GenerationFunction[] META_GEN_FUNCTION = null; + private static final boolean[] UNIQUE_META_FIELDS = null; + private static final int[] KEY_INDEXES = { 0 }; + private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR }; + private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR }); + private static final int TOTAL_NUM_OF_RECORDS = 10000; + private static final int RECORDS_PER_COMPONENT = 1000; + private static final int DATASET_ID = 101; + private static final String DATAVERSE_NAME = "TestDV"; + private static final String DATASET_NAME = "TestDS"; + private static final String DATA_TYPE_NAME = "DUMMY"; + private static final String NODE_GROUP_NAME = "DEFAULT"; + private static final StorageComponentProvider storageManager = new StorageComponentProvider(); + private static final int NUM_PARTITIONS = 2; + private static TestNodeController nc; + private static NCAppRuntimeContext ncAppCtx; + private static IDatasetLifecycleManager dsLifecycleMgr; + private static Dataset dataset; + private static ITransactionContext txnCtx; + private static TestLsmBtree[] primarylsmBtrees; + private static IHyracksTaskContext[] taskCtxs; + private static IIndexDataflowHelper[] indexDataflowHelpers; + private static LSMInsertDeleteOperatorNodePushable[] insertOps; + + @BeforeClass + public static void setUp() throws Exception { + System.out.println("SetUp: "); + TestHelper.deleteExistingInstanceFiles(); + nc = new TestNodeController(null, false); + nc.init(); + ncAppCtx = nc.getAppRuntimeContext(); + dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager(); + } + + @AfterClass + public static void tearDown() throws Exception { + System.out.println("TearDown"); + nc.deInit(); + TestHelper.deleteExistingInstanceFiles(); + } + + @Before + public void createIndex() throws Exception { + List<List<String>> partitioningKeys = new ArrayList<>(); + partitioningKeys.add(Collections.singletonList("key")); + dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME, + NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH, + partitioningKeys, null, null, null, false, null, false), + null, DatasetType.INTERNAL, DATASET_ID, 0); + taskCtxs = new IHyracksTaskContext[NUM_PARTITIONS]; + indexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS]; + primarylsmBtrees = new TestLsmBtree[NUM_PARTITIONS]; + insertOps = new LSMInsertDeleteOperatorNodePushable[NUM_PARTITIONS]; + JobId jobId = nc.newJobId(); + txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(jobId), + new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL)); + for (int i = 0; i < taskCtxs.length; i++) { + taskCtxs[i] = nc.createTestContext(jobId, i, false); + PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, + storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, i); + IndexDataflowHelperFactory iHelperFactory = + new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider()); + indexDataflowHelpers[i] = iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i); + indexDataflowHelpers[i].open(); + primarylsmBtrees[i] = (TestLsmBtree) indexDataflowHelpers[i].getIndexInstance(); + indexDataflowHelpers[i].close(); + insertOps[i] = nc.getInsertPipeline(taskCtxs[i], dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, + KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft(); + } + } + + @After + public void destroyIndex() throws Exception { + for (IIndexDataflowHelper indexDataflowHelper : indexDataflowHelpers) { + indexDataflowHelper.destroy(); + } + } + + @Test + public void testFlushOneFullOneEmpty() { + try { + // allow all operations + for (int i = 0; i < NUM_PARTITIONS; i++) { + ComponentRollbackTest.allowAllOps(primarylsmBtrees[i]); + } + + insertOps[0].open(); + TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS, + RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS); + VSizeFrame frame = new VSizeFrame(taskCtxs[0]); + FrameTupleAppender tupleAppender = new FrameTupleAppender(frame); + int numFlushes = 0; + for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) { + // flush every RECORDS_PER_COMPONENT records + if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) { + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOps[0], true); + } + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + numFlushes++; + } + ITupleReference tuple = tupleGenerator.next(); + DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOps[0]); + } + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOps[0], true); + } + insertOps[0].close(); + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + numFlushes++; + nc.getTransactionManager().commitTransaction(txnCtx.getTxnId()); + // search now and ensure partition 0 has all the records + ComponentRollbackTest.searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + // and that partition 1 has no records + ComponentRollbackTest.searchAndAssertCount(nc, 1, dataset, storageManager, 0); + // and that partition 0 has numFlushes disk components + Assert.assertEquals(numFlushes, primarylsmBtrees[0].getDiskComponents().size()); + // and that partition 1 has no disk components + Assert.assertEquals(0, primarylsmBtrees[1].getDiskComponents().size()); + } catch (Throwable e) { + e.printStackTrace(); + Assert.fail(e.getMessage()); + } + } + +} http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java new file mode 100644 index 0000000..77d3795 --- /dev/null +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java @@ -0,0 +1,266 @@ +/* + * 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.asterix.test.dataflow; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.asterix.app.bootstrap.TestNodeController; +import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo; +import org.apache.asterix.app.data.gen.TestTupleCounterFrameWriter; +import org.apache.asterix.app.data.gen.TupleGenerator; +import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction; +import org.apache.asterix.app.nc.NCAppRuntimeContext; +import org.apache.asterix.common.api.IDatasetLifecycleManager; +import org.apache.asterix.common.config.DatasetConfig.DatasetType; +import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable; +import org.apache.asterix.common.transactions.ITransactionContext; +import org.apache.asterix.common.transactions.ITransactionManager; +import org.apache.asterix.common.transactions.TransactionOptions; +import org.apache.asterix.external.util.DataflowUtils; +import org.apache.asterix.file.StorageComponentProvider; +import org.apache.asterix.metadata.entities.Dataset; +import org.apache.asterix.metadata.entities.Index; +import org.apache.asterix.metadata.entities.InternalDatasetDetails; +import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy; +import org.apache.asterix.om.types.ARecordType; +import org.apache.asterix.om.types.BuiltinType; +import org.apache.asterix.om.types.IAType; +import org.apache.asterix.test.common.TestHelper; +import org.apache.asterix.test.dataflow.ComponentRollbackTest.Searcher; +import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException; +import org.apache.hyracks.algebricks.runtime.base.IPushRuntime; +import org.apache.hyracks.api.comm.VSizeFrame; +import org.apache.hyracks.api.context.IHyracksTaskContext; +import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.api.job.JobId; +import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender; +import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference; +import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper; +import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory; +import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters; +import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree; +import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent; +import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor; +import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper; +import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class SearchCursorComponentSwitchTest { + private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 }; + private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" }, + new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false); + private static final GenerationFunction[] RECORD_GEN_FUNCTION = + { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC }; + private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false }; + private static final ARecordType META_TYPE = null; + private static final GenerationFunction[] META_GEN_FUNCTION = null; + private static final boolean[] UNIQUE_META_FIELDS = null; + private static final int[] KEY_INDEXES = { 0 }; + private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR }; + private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR }); + private static final int TOTAL_NUM_OF_RECORDS = 2000; + private static final int RECORDS_PER_COMPONENT = 1000; + private static final int DATASET_ID = 101; + private static final String DATAVERSE_NAME = "TestDV"; + private static final String DATASET_NAME = "TestDS"; + private static final String DATA_TYPE_NAME = "DUMMY"; + private static final String NODE_GROUP_NAME = "DEFAULT"; + private static final StorageComponentProvider storageManager = new StorageComponentProvider(); + private static TestNodeController nc; + private static TestLsmBtree lsmBtree; + private static NCAppRuntimeContext ncAppCtx; + private static IDatasetLifecycleManager dsLifecycleMgr; + private static Dataset dataset; + private static IHyracksTaskContext ctx; + private static IIndexDataflowHelper indexDataflowHelper; + private static ITransactionContext txnCtx; + private static LSMInsertDeleteOperatorNodePushable insertOp; + + @BeforeClass + public static void setUp() throws Exception { + System.out.println("SetUp: "); + TestHelper.deleteExistingInstanceFiles(); + nc = new TestNodeController(null, false); + nc.init(); + ncAppCtx = nc.getAppRuntimeContext(); + dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager(); + } + + @AfterClass + public static void tearDown() throws Exception { + System.out.println("TearDown"); + nc.deInit(); + TestHelper.deleteExistingInstanceFiles(); + } + + @Before + public void createIndex() throws Exception { + List<List<String>> partitioningKeys = new ArrayList<>(); + partitioningKeys.add(Collections.singletonList("key")); + dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME, + NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH, + partitioningKeys, null, null, null, false, null, false), + null, DatasetType.INTERNAL, DATASET_ID, 0); + PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, + storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0); + IndexDataflowHelperFactory iHelperFactory = + new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider()); + JobId jobId = nc.newJobId(); + ctx = nc.createTestContext(jobId, 0, false); + indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0); + indexDataflowHelper.open(); + lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance(); + indexDataflowHelper.close(); + txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx), + new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL)); + insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES, + KEY_INDICATORS_LIST, storageManager).getLeft(); + } + + @After + public void destroyIndex() throws Exception { + indexDataflowHelper.destroy(); + } + + void unblockSearch(TestLsmBtree lsmBtree) { + lsmBtree.addSearchCallback(sem -> sem.release()); + lsmBtree.allowSearch(1); + } + + @Test + public void testCursorSwitchSucceed() { + try { + // allow all operations + ComponentRollbackTest.allowAllOps(lsmBtree); + // except search + lsmBtree.clearSearchCallbacks(); + insertOp.open(); + TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS, + RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS); + VSizeFrame frame = new VSizeFrame(ctx); + FrameTupleAppender tupleAppender = new FrameTupleAppender(frame); + Searcher firstSearcher = null; + for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) { + // flush every RECORDS_PER_COMPONENT records + if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) { + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOp, true); + } + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + } + ITupleReference tuple = tupleGenerator.next(); + DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp); + } + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOp, true); + } + insertOp.close(); + // start the search + firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + // wait till firstSearcher enter the components + firstSearcher.waitUntilEntered(); + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + nc.getTransactionManager().commitTransaction(txnCtx.getTxnId()); + // unblock the search + unblockSearch(lsmBtree); + // ensure the search got the correct number + Assert.assertTrue(firstSearcher.result()); + // search now and ensure + searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + } catch (Throwable e) { + e.printStackTrace(); + Assert.fail(e.getMessage()); + } + } + + @Test + public void testCursorSwitchFails() { + try { + // allow all operations + ComponentRollbackTest.allowAllOps(lsmBtree); + // except search + lsmBtree.clearSearchCallbacks(); + insertOp.open(); + TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS, + RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS); + VSizeFrame frame = new VSizeFrame(ctx); + FrameTupleAppender tupleAppender = new FrameTupleAppender(frame); + Searcher firstSearcher = null; + for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) { + // flush every RECORDS_PER_COMPONENT records + if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) { + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOp, true); + } + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + } + ITupleReference tuple = tupleGenerator.next(); + DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp); + } + if (tupleAppender.getTupleCount() > 0) { + tupleAppender.write(insertOp, true); + } + insertOp.close(); + // start the search + firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS); + // wait till firstSearcher enter the components + firstSearcher.waitUntilEntered(); + dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false); + nc.getTransactionManager().commitTransaction(txnCtx.getTxnId()); + // merge all components + ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE); + List<ILSMDiskComponent> mergedComponents = new ArrayList<>(lsmBtree.getDiskComponents()); + BlockingIOOperationCallbackWrapper ioCallback = + new BlockingIOOperationCallbackWrapper(lsmBtree.getIOOperationCallback()); + mergeAccessor.scheduleMerge(ioCallback, mergedComponents); + ioCallback.waitForIO(); + // unblock the search + unblockSearch(lsmBtree); + // ensure the search got the correct number + Assert.assertTrue(firstSearcher.result()); + // search now and ensure + searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS); + } catch (Throwable e) { + e.printStackTrace(); + Assert.fail(e.getMessage()); + } + } + + private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset, + StorageComponentProvider storageManager, int numOfRecords) + throws HyracksDataException, AlgebricksException { + nc.newJobId(); + TestTupleCounterFrameWriter countOp = + ComponentRollbackTest.create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE), + Collections.emptyList(), Collections.emptyList(), false); + IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, + new NoMergePolicyFactory(), null, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager); + emptyTupleOp.open(); + emptyTupleOp.close(); + Assert.assertEquals(numOfRecords, countOp.getCount()); + } +} http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java index e0502de..0c4983a 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java @@ -21,7 +21,6 @@ package org.apache.asterix.test.dataflow; import java.util.Map; import org.apache.asterix.common.config.DatasetConfig.DatasetType; -import org.apache.asterix.common.context.DatasetLSMComponentIdGeneratorFactory; import org.apache.asterix.metadata.IDatasetDetails; import org.apache.asterix.metadata.declared.MetadataProvider; import org.apache.asterix.metadata.entities.Dataset; @@ -63,6 +62,6 @@ public class TestDataset extends Dataset { @Override public ILSMIOOperationCallbackFactory getIoOperationCallbackFactory(Index index) throws AlgebricksException { - return new TestLsmBtreeIoOpCallbackFactory(new DatasetLSMComponentIdGeneratorFactory(getDatasetId())); + return new TestLsmBtreeIoOpCallbackFactory(getComponentIdGeneratorFactory()); } } http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java index fa37c20..44967e3 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java @@ -18,8 +18,9 @@ */ package org.apache.asterix.test.dataflow; -import org.apache.asterix.common.ioopcallbacks.AbstractLSMIndexIOOperationCallbackFactory; import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback; +import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory; +import org.apache.hyracks.api.exceptions.HyracksDataException; import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator; import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory; import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent; @@ -28,7 +29,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex; import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent; -public class TestLsmBtreeIoOpCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory { +public class TestLsmBtreeIoOpCallbackFactory extends LSMBTreeIOOperationCallbackFactory { private static final long serialVersionUID = 1L; @@ -97,7 +98,8 @@ public class TestLsmBtreeIoOpCallbackFactory extends AbstractLSMIndexIOOperation } @Override - public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) { + public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) + throws HyracksDataException { super.afterFinalize(opType, newComponent); synchronized (TestLsmBtreeIoOpCallbackFactory.this) { if (newComponent != null) { http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java index 1cc24d5..1be1d26 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java @@ -54,6 +54,7 @@ import org.apache.asterix.transaction.management.service.logging.LogManager; import org.apache.asterix.transaction.management.service.recovery.AbstractCheckpointManager; import org.apache.hyracks.api.comm.VSizeFrame; import org.apache.hyracks.api.context.IHyracksTaskContext; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender; import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference; import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; @@ -126,9 +127,9 @@ public class CheckpointingTest { null, DatasetType.INTERNAL, DATASET_ID, 0); try { nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES, - KEY_INDICATOR_LIST); - IHyracksTaskContext ctx = nc.createTestContext(false); - nc.newJobId(); + KEY_INDICATOR_LIST, 0); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false); ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx), new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL)); // Prepare insert operation http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java index 5ee0e9f..cfd251b 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java @@ -28,6 +28,7 @@ import org.apache.asterix.test.runtime.LangExecutionUtil; import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.dataflow.ConnectorDescriptorId; import org.apache.hyracks.api.dataflow.TaskAttemptId; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.api.partitions.PartitionId; import org.apache.hyracks.comm.channels.NetworkOutputChannel; import org.apache.hyracks.control.nc.NodeControllerService; @@ -58,7 +59,8 @@ public class DeallocatableTest { final NodeControllerService ncs = (NodeControllerService) nc.getAppRuntimeContext().getServiceContext().getControllerService(); final TaskAttemptId taId = Mockito.mock(TaskAttemptId.class); - final IHyracksTaskContext ctx = nc.createTestContext(true); + JobId jobId = nc.newJobId(); + final IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true); final ConnectorDescriptorId codId = new ConnectorDescriptorId(1); final PartitionId pid = new PartitionId(ctx.getJobletContext().getJobId(), codId, 1, 1); final ChannelControlBlock ccb = ncs.getNetworkManager() http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java index cb83d56..8e8b3e9 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java @@ -33,7 +33,6 @@ import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction; import org.apache.asterix.common.config.DatasetConfig.DatasetType; import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable; import org.apache.asterix.common.exceptions.ExceptionUtils; -import org.apache.asterix.common.transactions.ITransactionContext; import org.apache.asterix.external.util.DataflowUtils; import org.apache.asterix.file.StorageComponentProvider; import org.apache.asterix.metadata.entities.Dataset; @@ -49,6 +48,7 @@ import org.apache.hyracks.api.comm.VSizeFrame; import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.exceptions.ErrorCode; import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender; import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference; import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; @@ -125,9 +125,9 @@ public class DiskIsFullTest { null, DatasetType.INTERNAL, DATASET_ID, 0); try { nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES, - KEY_INDICATOR_LIST); - IHyracksTaskContext ctx = nc.createTestContext(false); - nc.newJobId(); + KEY_INDICATOR_LIST, 0); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false); // Prepare insert operation LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft(); http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java index 7ced87d..cc6c0f7 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java @@ -18,8 +18,6 @@ */ package org.apache.asterix.test.storage; -import static org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption; - import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -51,9 +49,11 @@ import org.apache.hyracks.api.context.IHyracksTaskContext; import org.apache.hyracks.api.exceptions.ErrorCode; import org.apache.hyracks.api.exceptions.HyracksDataException; import org.apache.hyracks.api.io.FileSplit; +import org.apache.hyracks.api.job.JobId; import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider; import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper; import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory; +import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption; import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorNodePushable; import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; import org.junit.After; @@ -106,15 +106,15 @@ public class IndexDropOperatorNodePushableTest { Dataset dataset = new Dataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME, NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, InternalDatasetDetails.PartitioningStrategy.HASH, partitioningKeys, - null, null, null, false, null, false), null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID, - 0); + null, null, null, false, null, false), + null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID, 0); // create dataset - TestNodeController.PrimaryIndexInfo indexInfo = - nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES, - KEY_INDICATORS_LIST); + TestNodeController.PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, + META_TYPE, null, storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0); IndexDataflowHelperFactory helperFactory = new IndexDataflowHelperFactory(nc.getStorageManager(), indexInfo.getFileSplitProvider()); - IHyracksTaskContext ctx = nc.createTestContext(true); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true); IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0); dropInUse(ctx, helperFactory, dataflowHelper); dropInUseWithWait(ctx, helperFactory, dataflowHelper); @@ -144,21 +144,21 @@ public class IndexDropOperatorNodePushableTest { testExecutor.executeSqlppUpdateOrDdl("CREATE DATASET " + datasetName + "(KeyType) PRIMARY KEY id;", format); testExecutor.executeSqlppUpdateOrDdl("CREATE INDEX " + indexName + " on " + datasetName + "(foo)", format); final MetadataTransactionContext mdTxn = MetadataManager.INSTANCE.beginTransaction(); - ICcApplicationContext appCtx = - (ICcApplicationContext) ExecutionTestUtil.integrationUtil.getClusterControllerService() - .getApplicationContext(); + ICcApplicationContext appCtx = (ICcApplicationContext) ExecutionTestUtil.integrationUtil + .getClusterControllerService().getApplicationContext(); MetadataProvider metadataProver = new MetadataProvider(appCtx, null); metadataProver.setMetadataTxnContext(mdTxn); final String defaultDv = MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName(); final Dataset dataset = MetadataManager.INSTANCE.getDataset(mdTxn, defaultDv, datasetName); MetadataManager.INSTANCE.commitTransaction(mdTxn); - FileSplit[] splits = SplitsAndConstraintsUtil - .getIndexSplits(appCtx.getClusterStateManager(), dataset, indexName, Arrays.asList("asterix_nc1")); + FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset, + indexName, Arrays.asList("asterix_nc1")); final ConstantFileSplitProvider constantFileSplitProvider = new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1)); IndexDataflowHelperFactory helperFactory = new IndexDataflowHelperFactory(nc.getStorageManager(), constantFileSplitProvider); - IHyracksTaskContext ctx = nc.createTestContext(true); + JobId jobId = nc.newJobId(); + IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true); IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0); dropInUse(ctx, helperFactory, dataflowHelper); } finally { http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java index e445fe4..c33e2d1 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java @@ -34,6 +34,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex; import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent; import org.apache.hyracks.storage.am.lsm.common.impls.DiskComponentMetadata; +import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent; import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils; import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils; @@ -88,7 +89,7 @@ public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationC } @Override - public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) { + public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) throws HyracksDataException { // The operation was complete and the next I/O operation for the LSM index didn't start yet if (opType == LSMIOOperationType.FLUSH && newComponent != null) { synchronized (this) { @@ -100,6 +101,13 @@ public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationC } readIndex = (readIndex + 1) % mutableLastLSNs.length; } + if (newComponent == EmptyComponent.INSTANCE) { + // This component was just deleted, we refresh the component id, when it gets recycled, it will get + // the new id from the component id generator. + // It is assumed that the component delete caller will ensure that corresponding components in secondary + // indexes are deleted as well + idGenerator.refresh(); + } } } http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java ---------------------------------------------------------------------- diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java index 2d8eaed..a7468a7 100644 --- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java +++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java @@ -51,13 +51,25 @@ public abstract class AbstractOneInputOneOutputOneFramePushRuntime extends Abstr @Override public void close() throws HyracksDataException { + HyracksDataException closeException = null; try { flushIfNotFailed(); } catch (Exception e) { + closeException = HyracksDataException.create(e); writer.fail(); - throw e; } finally { - writer.close(); + try { + writer.close(); + } catch (Exception e) { + if (closeException == null) { + closeException = HyracksDataException.create(e); + } else { + closeException.addSuppressed(e); + } + } + } + if (closeException != null) { + throw closeException; } }