You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2018/05/21 03:35:25 UTC
[5/5] asterixdb git commit: [NO ISSUE][STO] Misc Storage Fixes and
Improvements
[NO ISSUE][STO] Misc Storage Fixes and Improvements
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
- This change introduces some improvements to storage
operations.
- Local RecoveryManager is now extensible.
- Bulk loaders now call the IO callback similar to
Flushes, making them less special and creating a
unified lifecycle for adding an index component.
- As a result, The IndexCheckpointManager doesn't need
to have a special treatment for components loaded
through the bulk load operation.
- Component Id have been added to the index checkpoint
files.
- Cleanup for the code of local recovery for failed flush
operations.
- Ensure that after local recovery of flushes, primary
and secondary indexes have the same index for mutable
memory component.
- The use of WAIT logs to ensure in-flight flushes
are scheduled didn't work as expected. A new log type
WAIT_FOR_FLUSHES was introduced to acheive the expected
behavior.
- The local test framework was made Extensible to support
more use cases.
- Test cases were added for component ids in checkpoint files.
The following scenarios were covered:
- Primary and secondary both have values when a flush is
shceduled.
- Primary have values but not secondary when a flush is
scheduled.
- Primary is empty and an index is created through bulk
load.
- Primary has a single component and secondary is created
through bulk load.
- Primary has multiple components and secondary is created
through bulk load.
- Each primary opTracker now keeps a list of ongoing flushes.
- FlushDataset now waits only for flushes only and
not all io operations.
- Previously, we had many flushes scheduled on open datasets.
This was not detected but after this change, a failure
is thrown in such cases.
- Flush operations dont need to extend the comparable
interface anymore since they are FIFO per index.
Change-Id: If24c9baaac2b79e7d1acf47fa2601767388ce988
Reviewed-on: https://asterix-gerrit.ics.uci.edu/2632
Sonar-Qube: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/10a3f21d
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/10a3f21d
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/10a3f21d
Branch: refs/heads/master
Commit: 10a3f21dee373886e68239c61dd828323d90f0ec
Parents: 1a6de0b
Author: Abdullah Alamoudi <ba...@gmail.com>
Authored: Mon May 21 02:32:55 2018 +0300
Committer: abdullah alamoudi <ba...@gmail.com>
Committed: Sun May 20 20:33:40 2018 -0700
----------------------------------------------------------------------
.../asterix/app/nc/IndexCheckpointManager.java | 75 ++-
.../app/nc/IndexCheckpointManagerProvider.java | 7 +-
.../asterix/app/nc/NCAppRuntimeContext.java | 5 +-
.../apache/asterix/app/nc/RecoveryManager.java | 47 +-
.../asterix/app/nc/TransactionSubsystem.java | 5 +-
.../hyracks/bootstrap/NCApplication.java | 8 +-
.../common/AsterixHyracksIntegrationUtil.java | 10 +-
.../app/bootstrap/TestNodeController.java | 100 +++-
.../data/gen/ABooleanFieldValueGenerator.java | 2 +-
.../data/gen/ADoubleFieldValueGenerator.java | 2 +-
.../app/data/gen/AInt32FieldValueGenerator.java | 2 +-
.../app/data/gen/AInt64FieldValueGenerator.java | 2 +-
.../app/data/gen/ARecordValueGenerator.java | 2 +-
.../data/gen/AStringFieldValueGenerator.java | 2 +-
.../app/data/gen/RecordTupleGenerator.java | 126 ++++
.../asterix/app/data/gen/TupleGenerator.java | 126 ----
.../test/dataflow/AInt32ValueGenerator.java | 36 ++
.../test/dataflow/AInt64ValueGenerator.java | 36 ++
.../CheckpointInSecondaryIndexTest.java | 577 +++++++++++++++++++
.../test/dataflow/ComponentRollbackTest.java | 20 +-
.../test/dataflow/IoCallbackFailureTest.java | 4 +-
.../test/dataflow/LSMFlushRecoveryTest.java | 6 +-
.../asterix/test/dataflow/LogMarkerTest.java | 32 +-
.../dataflow/MultiPartitionLSMIndexTest.java | 8 +-
.../SearchCursorComponentSwitchTest.java | 12 +-
.../asterix/test/dataflow/StorageTestUtils.java | 50 +-
.../asterix/test/dataflow/TestDataset.java | 19 +
.../test/dataflow/TransactionAbortTest.java | 4 +-
.../asterix/test/logging/CheckpointingTest.java | 42 +-
.../asterix/test/runtime/ExecutionTestUtil.java | 5 +-
.../common/api/INcApplicationContext.java | 4 +-
.../asterix/common/context/DatasetInfo.java | 2 +-
.../common/context/DatasetLifecycleManager.java | 96 ++-
.../context/PrimaryIndexOperationTracker.java | 39 +-
.../asterix/common/dataflow/LSMIndexUtil.java | 14 +
.../ioopcallbacks/LSMIOOperationCallback.java | 93 ++-
.../common/storage/IIndexCheckpointManager.java | 40 +-
.../asterix/common/storage/IndexCheckpoint.java | 15 +-
.../transactions/IRecoveryManagerFactory.java | 35 ++
.../asterix/common/transactions/LogRecord.java | 3 +-
.../asterix/common/transactions/LogType.java | 4 +
.../asterix/common/utils/StoragePathUtil.java | 16 +-
.../LSMIOOperationCallbackTest.java | 21 +-
.../ioopcallbacks/TestLSMIndexAccessor.java | 2 +-
...ernalFilesIndexCreateOperatorDescriptor.java | 13 +-
...ilesIndexModificationOperatorDescriptor.java | 5 +-
...ternalIndexBulkLoadOperatorNodePushable.java | 14 +
...rnalIndexBulkModifyOperatorNodePushable.java | 7 +-
.../CheckpointPartitionIndexesTask.java | 26 +-
.../messaging/MarkComponentValidTask.java | 24 +-
.../messaging/ReplicateFileTask.java | 4 +-
.../replication/sync/IndexSynchronizer.java | 22 +-
.../LSMIndexBulkLoadOperatorNodePushable.java | 29 +-
.../LSMSecondaryIndexBulkLoadNodePushable.java | 15 +-
.../std/FlushDatasetOperatorDescriptor.java | 9 +-
.../PersistentLocalResourceRepository.java | 24 +-
.../management/service/logging/LogBuffer.java | 23 +-
.../management/service/logging/LogManager.java | 29 +-
.../logging/LogManagerWithReplication.java | 14 +-
.../am/common/api/ITwoPCIndexBulkLoader.java | 2 +-
.../am/lsm/btree/impls/ExternalBTree.java | 81 +--
.../lsm/btree/impls/ExternalBTreeWithBuddy.java | 88 +--
.../lsm/btree/impls/LSMBTreeFlushOperation.java | 4 +-
.../am/lsm/common/api/ILSMComponent.java | 5 +
.../am/lsm/common/api/ILSMComponentId.java | 10 +
.../storage/am/lsm/common/api/ILSMHarness.java | 4 +-
.../am/lsm/common/api/ILSMIOOperation.java | 13 +
.../storage/am/lsm/common/api/ILSMIndex.java | 28 +-
.../am/lsm/common/api/ILSMIndexAccessor.java | 4 +-
.../storage/am/lsm/common/api/ITwoPCIndex.java | 5 +-
.../common/api/IoOperationCompleteListener.java | 29 +
.../am/lsm/common/api/LSMOperationType.java | 1 +
.../lsm/common/impls/AbstractIoOperation.java | 30 +
.../lsm/common/impls/AbstractLSMComponent.java | 5 +
.../am/lsm/common/impls/AbstractLSMIndex.java | 56 +-
.../impls/AbstractLSMMemoryComponent.java | 4 +-
.../ChainedLSMDiskComponentBulkLoader.java | 12 +-
.../am/lsm/common/impls/EmptyComponent.java | 7 +-
.../lsm/common/impls/ExternalIndexHarness.java | 8 +-
.../am/lsm/common/impls/FlushOperation.java | 11 +-
.../am/lsm/common/impls/LSMComponentId.java | 8 +-
.../storage/am/lsm/common/impls/LSMHarness.java | 37 +-
.../impls/LSMIndexDiskComponentBulkLoader.java | 26 +-
.../lsm/common/impls/LSMTreeIndexAccessor.java | 4 +-
.../am/lsm/common/impls/LoadOperation.java | 19 +-
.../am/lsm/common/impls/NoOpIoOperation.java | 13 +
.../am/lsm/common/impls/TracedIOOperation.java | 37 +-
.../am/lsm/common/util/LSMComponentIdUtils.java | 7 +-
.../impls/LSMInvertedIndexAccessor.java | 4 +-
.../am/lsm/rtree/impls/ExternalRTree.java | 146 ++---
90 files changed, 1981 insertions(+), 741 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
index a012f1e..c1a81a3 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
@@ -22,6 +22,7 @@ import java.io.BufferedWriter;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -54,29 +55,36 @@ public class IndexCheckpointManager implements IIndexCheckpointManager {
}
@Override
- public synchronized void init(long lsn) throws HyracksDataException {
- final List<IndexCheckpoint> checkpoints = getCheckpoints();
+ public synchronized void init(String lastComponentTimestamp, long lsn) throws HyracksDataException {
+ List<IndexCheckpoint> checkpoints;
+ try {
+ checkpoints = getCheckpoints();
+ } catch (ClosedByInterruptException e) {
+ throw HyracksDataException.create(e);
+ }
if (!checkpoints.isEmpty()) {
LOGGER.warn(() -> "Checkpoints found on initializing: " + indexPath);
delete();
}
- IndexCheckpoint firstCheckpoint = IndexCheckpoint.first(lsn);
+ IndexCheckpoint firstCheckpoint = IndexCheckpoint.first(lastComponentTimestamp, lsn);
persist(firstCheckpoint);
}
@Override
- public synchronized void replicated(String componentTimestamp, long masterLsn) throws HyracksDataException {
+ public synchronized void replicated(String componentTimestamp, long masterLsn, long componentId)
+ throws HyracksDataException {
final Long localLsn = getLatest().getMasterNodeFlushMap().get(masterLsn);
if (localLsn == null) {
throw new IllegalStateException("Component flushed before lsn mapping was received");
}
- flushed(componentTimestamp, localLsn);
+ flushed(componentTimestamp, localLsn, componentId);
}
@Override
- public synchronized void flushed(String componentTimestamp, long lsn) throws HyracksDataException {
+ public synchronized void flushed(String componentTimestamp, long lsn, long componentId)
+ throws HyracksDataException {
final IndexCheckpoint latest = getLatest();
- IndexCheckpoint nextCheckpoint = IndexCheckpoint.next(latest, lsn, componentTimestamp);
+ IndexCheckpoint nextCheckpoint = IndexCheckpoint.next(latest, lsn, componentTimestamp, componentId);
persist(nextCheckpoint);
deleteHistory(nextCheckpoint.getId(), HISTORY_CHECKPOINTS);
}
@@ -85,19 +93,19 @@ public class IndexCheckpointManager implements IIndexCheckpointManager {
public synchronized void masterFlush(long masterLsn, long localLsn) throws HyracksDataException {
final IndexCheckpoint latest = getLatest();
latest.getMasterNodeFlushMap().put(masterLsn, localLsn);
- final IndexCheckpoint next =
- IndexCheckpoint.next(latest, latest.getLowWatermark(), latest.getValidComponentTimestamp());
+ final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(),
+ latest.getValidComponentTimestamp(), latest.getLastComponentId());
persist(next);
notifyAll();
}
@Override
- public synchronized long getLowWatermark() {
+ public synchronized long getLowWatermark() throws HyracksDataException {
return getLatest().getLowWatermark();
}
@Override
- public synchronized boolean isFlushed(long masterLsn) {
+ public synchronized boolean isFlushed(long masterLsn) throws HyracksDataException {
if (masterLsn == BULKLOAD_LSN) {
return true;
}
@@ -110,18 +118,28 @@ public class IndexCheckpointManager implements IIndexCheckpointManager {
}
@Override
- public Optional<String> getValidComponentTimestamp() {
- final String validComponentTimestamp = getLatest().getValidComponentTimestamp();
+ public Optional<String> getValidComponentTimestamp() throws HyracksDataException {
+ String validComponentTimestamp = getLatest().getValidComponentTimestamp();
return validComponentTimestamp != null ? Optional.of(validComponentTimestamp) : Optional.empty();
}
@Override
- public int getCheckpointCount() {
- return getCheckpoints().size();
+ public int getCheckpointCount() throws HyracksDataException {
+ try {
+ return getCheckpoints().size();
+ } catch (ClosedByInterruptException e) {
+ throw HyracksDataException.create(e);
+ }
}
- private IndexCheckpoint getLatest() {
- final List<IndexCheckpoint> checkpoints = getCheckpoints();
+ @Override
+ public synchronized IndexCheckpoint getLatest() throws HyracksDataException {
+ List<IndexCheckpoint> checkpoints;
+ try {
+ checkpoints = getCheckpoints();
+ } catch (ClosedByInterruptException e) {
+ throw HyracksDataException.create(e);
+ }
if (checkpoints.isEmpty()) {
throw new IllegalStateException("Couldn't find any checkpoints for resource: " + indexPath);
}
@@ -129,13 +147,34 @@ public class IndexCheckpointManager implements IIndexCheckpointManager {
return checkpoints.get(0);
}
- private List<IndexCheckpoint> getCheckpoints() {
+ @Override
+ public synchronized void setLastComponentId(long componentId) throws HyracksDataException {
+ final IndexCheckpoint latest = getLatest();
+ final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(),
+ latest.getValidComponentTimestamp(), componentId);
+ persist(next);
+ }
+
+ @Override
+ public synchronized void advanceValidComponentTimestamp(String timestamp) throws HyracksDataException {
+ final IndexCheckpoint latest = getLatest();
+ if (latest.getValidComponentTimestamp() == null
+ || timestamp.compareTo(latest.getValidComponentTimestamp()) > 0) {
+ final IndexCheckpoint next =
+ IndexCheckpoint.next(latest, latest.getLowWatermark(), timestamp, latest.getLastComponentId());
+ persist(next);
+ }
+ }
+
+ private List<IndexCheckpoint> getCheckpoints() throws ClosedByInterruptException {
List<IndexCheckpoint> checkpoints = new ArrayList<>();
final File[] checkpointFiles = indexPath.toFile().listFiles(CHECKPOINT_FILE_FILTER);
if (checkpointFiles != null) {
for (File checkpointFile : checkpointFiles) {
try {
checkpoints.add(read(checkpointFile.toPath()));
+ } catch (ClosedByInterruptException e) {
+ throw e;
} catch (IOException e) {
LOGGER.warn(() -> "Couldn't read index checkpoint file: " + checkpointFile, e);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManagerProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManagerProvider.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManagerProvider.java
index 19ad8f6..e0b3105 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManagerProvider.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManagerProvider.java
@@ -25,6 +25,7 @@ import java.util.Map;
import org.apache.asterix.common.storage.IIndexCheckpointManager;
import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.asterix.common.storage.ResourceReference;
+import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.IIOManager;
@@ -53,14 +54,10 @@ public class IndexCheckpointManagerProvider implements IIndexCheckpointManagerPr
private IndexCheckpointManager create(ResourceReference ref) {
try {
- final Path indexPath = getIndexPath(ref);
+ final Path indexPath = StoragePathUtil.getIndexPath(ioManager, ref);
return new IndexCheckpointManager(indexPath);
} catch (HyracksDataException e) {
throw new IllegalStateException(e);
}
}
-
- private Path getIndexPath(ResourceReference indexRef) throws HyracksDataException {
- return ioManager.resolve(indexRef.getRelativePath().toString()).getFile().toPath();
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index c3b1bae..c3201c3 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -60,6 +60,7 @@ import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
import org.apache.asterix.common.storage.IReplicaManager;
import org.apache.asterix.common.transactions.IRecoveryManager;
import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
+import org.apache.asterix.common.transactions.IRecoveryManagerFactory;
import org.apache.asterix.common.transactions.ITransactionSubsystem;
import org.apache.asterix.external.library.ExternalLibraryManager;
import org.apache.asterix.file.StorageComponentProvider;
@@ -168,7 +169,7 @@ public class NCAppRuntimeContext implements INcApplicationContext {
}
@Override
- public void initialize(boolean initialRun) throws IOException {
+ public void initialize(IRecoveryManagerFactory recoveryManagerFactory, boolean initialRun) throws IOException {
ioManager = getServiceContext().getIoManager();
threadExecutor =
MaintainedThreadNameExecutorService.newCachedThreadPool(getServiceContext().getThreadFactory());
@@ -183,7 +184,7 @@ public class NCAppRuntimeContext implements INcApplicationContext {
new PersistentLocalResourceRepositoryFactory(ioManager, indexCheckpointManagerProvider);
localResourceRepository =
(PersistentLocalResourceRepository) persistentLocalResourceRepositoryFactory.createRepository();
- txnSubsystem = new TransactionSubsystem(this);
+ txnSubsystem = new TransactionSubsystem(this, recoveryManagerFactory);
IRecoveryManager recoveryMgr = txnSubsystem.getRecoveryManager();
SystemState systemState = recoveryMgr.getSystemState();
if (initialRun || systemState == SystemState.PERMANENT_DATA_LOSS) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
index efd173f..5e8a5e8 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
@@ -77,6 +77,8 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId.IdCompareResult;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
@@ -102,14 +104,14 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
private static final String RECOVERY_FILES_DIR_NAME = "recovery_temp";
private Map<Long, JobEntityCommits> jobId2WinnerEntitiesMap = null;
private final long cachedEntityCommitsPerJobSize;
- private final PersistentLocalResourceRepository localResourceRepository;
+ protected final PersistentLocalResourceRepository localResourceRepository;
private final ICheckpointManager checkpointManager;
private SystemState state;
- private final INCServiceContext serviceCtx;
- private final INcApplicationContext appCtx;
+ protected final INCServiceContext serviceCtx;
+ protected final INcApplicationContext appCtx;
private static final TxnId recoveryTxnId = new TxnId(-1);
- public RecoveryManager(ITransactionSubsystem txnSubsystem, INCServiceContext serviceCtx) {
+ public RecoveryManager(INCServiceContext serviceCtx, ITransactionSubsystem txnSubsystem) {
this.serviceCtx = serviceCtx;
this.txnSubsystem = txnSubsystem;
this.appCtx = txnSubsystem.getApplicationContext();
@@ -225,6 +227,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
break;
case LogType.FLUSH:
case LogType.WAIT:
+ case LogType.WAIT_FOR_FLUSHES:
case LogType.MARKER:
case LogType.FILTER:
break;
@@ -392,7 +395,6 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
logRecord = logReader.next();
continue;
}
- idGenerator.refresh();
DatasetInfo dsInfo = datasetLifecycleManager.getDatasetInfo(datasetId);
// we only need to flush open indexes here (opened by previous update records)
// if an index has no ongoing updates, then it's memory component must be empty
@@ -401,23 +403,15 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
if (iInfo.isOpen() && iInfo.getPartition() == partition) {
maxDiskLastLsn = resourceId2MaxLSNMap.get(iInfo.getResourceId());
index = iInfo.getIndex();
- LSMIOOperationCallback ioCallback =
- (LSMIOOperationCallback) index.getIOOperationCallback();
if (logRecord.getLSN() > maxDiskLastLsn
&& !index.isCurrentMutableComponentEmpty()) {
// schedule flush
- redoFlush(index, logRecord, idGenerator.getId());
+ redoFlush(index, logRecord);
redoCount++;
} else {
- if (index.isMemoryComponentsAllocated()) {
- // if the memory component has been allocated, we
- // force it to receive the same Id
- index.getCurrentMemoryComponent().resetId(idGenerator.getId(), true);
- } else {
- // otherwise, we refresh the id stored in ioCallback
- // to ensure the memory component receives correct Id upon activation
- ioCallback.forceRefreshNextId(idGenerator.getId());
- }
+ // otherwise, do nothing since this component had no records when flush was
+ // scheduled.. TODO: update checkpoint file? and do the
+ // lsn checks from the checkpoint file
}
}
}
@@ -427,6 +421,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
case LogType.ENTITY_COMMIT:
case LogType.ABORT:
case LogType.WAIT:
+ case LogType.WAIT_FOR_FLUSHES:
case LogType.MARKER:
//do nothing
break;
@@ -683,6 +678,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
case LogType.FLUSH:
case LogType.FILTER:
case LogType.WAIT:
+ case LogType.WAIT_FOR_FLUSHES:
case LogType.MARKER:
//ignore
break;
@@ -822,8 +818,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
}
}
- private static void redoFlush(ILSMIndex index, ILogRecord logRecord, ILSMComponentId nextId)
- throws HyracksDataException {
+ private static void redoFlush(ILSMIndex index, ILogRecord logRecord) throws HyracksDataException {
long flushLsn = logRecord.getLSN();
Map<String, Object> flushMap = new HashMap<>();
flushMap.put(LSMIOOperationCallback.KEY_FLUSH_LOG_LSN, flushLsn);
@@ -832,7 +827,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
long minId = logRecord.getFlushingComponentMinId();
long maxId = logRecord.getFlushingComponentMaxId();
ILSMComponentId id = new LSMComponentId(minId, maxId);
- flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, nextId);
+ flushMap.put(LSMIOOperationCallback.KEY_NEXT_COMPONENT_ID, index.getCurrentMemoryComponent().getId());
if (!index.getDiskComponents().isEmpty()) {
ILSMDiskComponent diskComponent = index.getDiskComponents().get(0);
ILSMComponentId maxDiskComponentId = diskComponent.getId();
@@ -842,7 +837,17 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
}
}
index.getCurrentMemoryComponent().resetId(id, true);
- accessor.scheduleFlush();
+ ILSMIOOperation flush = accessor.scheduleFlush();
+ try {
+ flush.sync();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw HyracksDataException.create(e);
+ }
+ if (flush.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flush.getFailure());
+ }
+ index.resetCurrentComponentIndex();
}
private class JobEntityCommits {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/TransactionSubsystem.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/TransactionSubsystem.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/TransactionSubsystem.java
index 8158096..79c87c0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/TransactionSubsystem.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/TransactionSubsystem.java
@@ -30,6 +30,7 @@ import org.apache.asterix.common.transactions.ICheckpointManager;
import org.apache.asterix.common.transactions.ILockManager;
import org.apache.asterix.common.transactions.ILogManager;
import org.apache.asterix.common.transactions.IRecoveryManager;
+import org.apache.asterix.common.transactions.IRecoveryManagerFactory;
import org.apache.asterix.common.transactions.ITransactionManager;
import org.apache.asterix.common.transactions.ITransactionSubsystem;
import org.apache.asterix.transaction.management.service.locking.ConcurrentLockManager;
@@ -59,7 +60,7 @@ public class TransactionSubsystem implements ITransactionSubsystem {
private long profilerEntityCommitLogCount = 0;
private EntityCommitProfiler ecp;
- public TransactionSubsystem(INcApplicationContext appCtx) {
+ public TransactionSubsystem(INcApplicationContext appCtx, IRecoveryManagerFactory recoveryManagerFactory) {
this.appCtx = appCtx;
this.id = appCtx.getServiceContext().getNodeId();
this.txnProperties = appCtx.getTransactionProperties();
@@ -78,7 +79,7 @@ public class TransactionSubsystem implements ITransactionSubsystem {
}
this.logManager = replicationEnabled ? new LogManagerWithReplication(this) : new LogManager(this);
- this.recoveryManager = new RecoveryManager(this, appCtx.getServiceContext());
+ this.recoveryManager = recoveryManagerFactory.createRecoveryManager(appCtx.getServiceContext(), this);
if (txnProperties.isCommitProfilerEnabled()) {
ecp = new EntityCommitProfiler(this, this.txnProperties.getCommitProfilerReportInterval());
((ExecutorService) appCtx.getThreadExecutor()).submit(ecp);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
index 494198b..b8c7e15 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.apache.asterix.api.http.server.ServletConstants;
import org.apache.asterix.api.http.server.StorageApiServlet;
import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.app.nc.RecoveryManager;
import org.apache.asterix.app.replication.message.RegistrationTasksRequestMessage;
import org.apache.asterix.common.api.AsterixThreadFactory;
import org.apache.asterix.common.api.INcApplicationContext;
@@ -44,6 +45,7 @@ import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.transactions.Checkpoint;
import org.apache.asterix.common.transactions.IRecoveryManager;
import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
+import org.apache.asterix.common.transactions.IRecoveryManagerFactory;
import org.apache.asterix.common.utils.PrintUtil;
import org.apache.asterix.common.utils.Servlets;
import org.apache.asterix.common.utils.StorageConstants;
@@ -121,7 +123,7 @@ public class NCApplication extends BaseNCApplication {
}
updateOnNodeJoin();
}
- runtimeContext.initialize(runtimeContext.getNodeProperties().isInitialRun());
+ runtimeContext.initialize(getRecoveryManagerFactory(), runtimeContext.getNodeProperties().isInitialRun());
MessagingProperties messagingProperties = runtimeContext.getMessagingProperties();
IMessageBroker messageBroker = new NCMessageBroker(controllerService, messagingProperties);
this.ncServiceCtx.setMessageBroker(messageBroker);
@@ -144,6 +146,10 @@ public class NCApplication extends BaseNCApplication {
performLocalCleanUp();
}
+ protected IRecoveryManagerFactory getRecoveryManagerFactory() {
+ return RecoveryManager::new;
+ }
+
@Override
protected void configureLoggingLevel(Level level) {
super.configureLoggingLevel(level);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
index cc70d57..4b245a8 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
@@ -97,7 +97,8 @@ public class AsterixHyracksIntegrationUtil {
* main method to run a simple 2 node cluster in-process
* suggested VM arguments: <code>-enableassertions -Xmx2048m -Dfile.encoding=UTF-8</code>
*
- * @param args unused
+ * @param args
+ * unused
*/
public static void main(String[] args) throws Exception {
TestUtils.redirectLoggingToConsole();
@@ -226,11 +227,14 @@ public class AsterixHyracksIntegrationUtil {
return ncConfig;
}
- protected INCApplication createNCApplication() {
+ protected INCApplication createNCApplication()
+ throws InstantiationException, IllegalAccessException, ClassNotFoundException {
+ // Instead of using this flag, RecoveryManagerTest should set the desired class in its config file
if (!gracefulShutdown) {
return new UngracefulShutdownNCApplication();
}
- return new NCApplication();
+ String ncAppClass = (String) configManager.get(NCConfig.Option.APP_CLASS);
+ return (INCApplication) Class.forName(ncAppClass).newInstance();
}
private NCConfig fixupIODevices(NCConfig ncConfig) throws IOException, AsterixException, CmdLineException {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 5cda9f2..5be349e 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
@@ -37,7 +37,6 @@ import org.apache.asterix.common.context.TransactionSubsystemProvider;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
import org.apache.asterix.common.transactions.ITransactionManager;
import org.apache.asterix.common.transactions.TxnId;
import org.apache.asterix.dataflow.data.nontagged.MissingWriterFactory;
@@ -58,17 +57,18 @@ import org.apache.asterix.om.types.IAType;
import org.apache.asterix.runtime.formats.FormatUtils;
import org.apache.asterix.runtime.formats.NonTaggedDataFormat;
import org.apache.asterix.runtime.job.listener.JobEventListenerFactory;
+import org.apache.asterix.runtime.operators.LSMIndexBulkLoadOperatorDescriptor.BulkLoadUsage;
+import org.apache.asterix.runtime.operators.LSMIndexBulkLoadOperatorNodePushable;
import org.apache.asterix.runtime.operators.LSMPrimaryUpsertOperatorNodePushable;
import org.apache.asterix.runtime.utils.CcApplicationContext;
import org.apache.asterix.test.runtime.ExecutionTestUtil;
-import org.apache.asterix.transaction.management.opcallbacks.AbstractIndexModificationOperationCallback.Operation;
-import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexModificationOperationCallbackFactory;
import org.apache.asterix.transaction.management.runtime.CommitRuntime;
import org.apache.asterix.transaction.management.service.logging.LogReader;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.operators.base.SinkRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.operators.std.AssignRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.operators.std.EmptyTupleSourceRuntimeFactory;
import org.apache.hyracks.api.comm.IFrameWriter;
@@ -166,7 +166,7 @@ public class TestNodeController {
public void deInit(boolean cleanupOnStop) throws Exception {
ExternalUDFLibrarian.removeLibraryDir();
- ExecutionTestUtil.tearDown(cleanupOnStop);
+ ExecutionTestUtil.tearDown(cleanupOnStop, runHDFS);
}
public void setOpts(List<Pair<IOption, Object>> opts) {
@@ -186,7 +186,7 @@ public class TestNodeController {
return new TxnId(jobId.getId());
}
- public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx,
+ public Pair<LSMInsertDeleteOperatorNodePushable, IPushRuntime> getInsertPipeline(IHyracksTaskContext ctx,
Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType, int[] filterFields,
int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators,
StorageComponentProvider storageComponentProvider, Index secondaryIndex)
@@ -195,7 +195,39 @@ public class TestNodeController {
primaryKeyIndicators, storageComponentProvider, secondaryIndex, IndexOperation.INSERT);
}
- public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx,
+ public Pair<SecondaryIndexInfo, LSMIndexBulkLoadOperatorNodePushable> getBulkLoadSecondaryOperator(
+ IHyracksTaskContext ctx, Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType,
+ ARecordType metaType, int[] filterFields, int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators,
+ StorageComponentProvider storageComponentProvider, Index secondaryIndex, int numElementsHint)
+ throws HyracksDataException, RemoteException, ACIDException, AlgebricksException {
+ try {
+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+ org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
+ DatasetUtil.getMergePolicyFactory(dataset, mdTxnCtx);
+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+ PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+ mergePolicy.first, mergePolicy.second, filterFields, primaryKeyIndexes, primaryKeyIndicators);
+ SecondaryIndexInfo secondaryIndexInfo = new SecondaryIndexInfo(primaryIndexInfo, secondaryIndex);
+ IIndexDataflowHelperFactory secondaryIndexHelperFactory = new IndexDataflowHelperFactory(
+ storageComponentProvider.getStorageManager(), secondaryIndexInfo.fileSplitProvider);
+ IIndexDataflowHelperFactory primaryIndexHelperFactory = new IndexDataflowHelperFactory(
+ storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+ int[] fieldPermutation = new int[secondaryIndex.getKeyFieldNames().size()];
+ for (int i = 0; i < fieldPermutation.length; i++) {
+ fieldPermutation[i] = i;
+ }
+ LSMIndexBulkLoadOperatorNodePushable op =
+ new LSMIndexBulkLoadOperatorNodePushable(secondaryIndexHelperFactory, primaryIndexHelperFactory,
+ ctx, 0, fieldPermutation, 1.0F, false, numElementsHint, true, secondaryIndexInfo.rDesc,
+ BulkLoadUsage.CREATE_INDEX, dataset.getDatasetId());
+ op.setOutputFrameWriter(0, new SinkRuntimeFactory().createPushRuntime(ctx)[0], null);
+ return Pair.of(secondaryIndexInfo, op);
+ } catch (Throwable th) {
+ throw HyracksDataException.create(th);
+ }
+ }
+
+ public Pair<LSMInsertDeleteOperatorNodePushable, IPushRuntime> getInsertPipeline(IHyracksTaskContext ctx,
Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType, int[] filterFields,
int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators,
StorageComponentProvider storageComponentProvider, Index secondaryIndex, IndexOperation op)
@@ -210,10 +242,8 @@ public class TestNodeController {
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
mergePolicy.first, mergePolicy.second, filterFields, primaryKeyIndexes, primaryKeyIndicators);
- IModificationOperationCallbackFactory modOpCallbackFactory =
- new PrimaryIndexModificationOperationCallbackFactory(dataset.getDatasetId(),
- primaryIndexInfo.primaryKeyIndexes, TXN_SUBSYSTEM_PROVIDER, Operation.get(op),
- ResourceType.LSM_BTREE);
+ IModificationOperationCallbackFactory modOpCallbackFactory = dataset.getModificationCallbackFactory(
+ storageComponentProvider, primaryIndexInfo.index, op, primaryIndexInfo.primaryKeyIndexes);
IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
RecordDescriptor recordDesc =
recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0);
@@ -272,16 +302,18 @@ public class TestNodeController {
secondaryIndexInfo.insertFieldsPermutations, secondaryIndexInfo.rDesc, op, false,
secondaryIndexHelperFactory, secondaryModCallbackFactory, null);
assignOp.setOutputFrameWriter(0, secondaryInsertOp, secondaryIndexInfo.rDesc);
- CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
- secondaryIndexInfo.primaryKeyIndexes, true, ctx.getTaskAttemptId().getTaskId().getPartition(),
- true);
+
+ IPushRuntime commitOp =
+ dataset.getCommitRuntimeFactory(mdProvider, secondaryIndexInfo.primaryKeyIndexes, true)
+ .createPushRuntime(ctx)[0];
+
secondaryInsertOp.setOutputFrameWriter(0, commitOp, secondaryIndexInfo.rDesc);
commitOp.setInputRecordDescriptor(0, secondaryIndexInfo.rDesc);
return Pair.of(insertOp, commitOp);
} else {
- CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
- primaryIndexInfo.primaryKeyIndexes, true, ctx.getTaskAttemptId().getTaskId().getPartition(),
- true);
+ IPushRuntime commitOp =
+ dataset.getCommitRuntimeFactory(mdProvider, primaryIndexInfo.primaryKeyIndexes, true)
+ .createPushRuntime(ctx)[0];
insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
return Pair.of(insertOp, commitOp);
@@ -380,7 +412,6 @@ public class TestNodeController {
(ICcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext(), dataverse);
SecondaryIndexInfo secondaryIndexInfo = new SecondaryIndexInfo(primaryIndexInfo, secondaryIndex);
try {
-
IResourceFactory resourceFactory = primaryIndexInfo.dataset.getResourceFactory(mdProvider, secondaryIndex,
primaryIndexInfo.recordType, primaryIndexInfo.metaType, mergePolicy.first, mergePolicy.second);
IndexBuilderFactory indexBuilderFactory =
@@ -419,9 +450,9 @@ public class TestNodeController {
secondaryIndexSerdes[i] =
SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(secondaryKeyTypes[i]);
}
- for (; i < primaryKeyTypes.length; i++) {
- secondaryIndexSerdes[i] =
- SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(primaryKeyTypes[i]);
+ for (; i < secondaryKeyTypes.length + primaryKeyTypes.length; i++) {
+ secondaryIndexSerdes[i] = SerializerDeserializerProvider.INSTANCE
+ .getSerializerDeserializer(primaryKeyTypes[i - secondaryKeyTypes.length]);
}
return secondaryIndexSerdes;
}
@@ -447,8 +478,9 @@ public class TestNodeController {
for (; i < secondaryKeyTypes.length; i++) {
secondaryIndexTypeTraits[i] = TypeTraitProvider.INSTANCE.getTypeTrait(secondaryKeyTypes[i]);
}
- for (; i < primaryKeyTypes.length; i++) {
- secondaryIndexTypeTraits[i] = TypeTraitProvider.INSTANCE.getTypeTrait(primaryKeyTypes[i]);
+ for (; i < secondaryKeyTypes.length + primaryKeyTypes.length; i++) {
+ secondaryIndexTypeTraits[i] =
+ TypeTraitProvider.INSTANCE.getTypeTrait(primaryKeyTypes[i - secondaryKeyTypes.length]);
}
return secondaryIndexTypeTraits;
}
@@ -491,12 +523,14 @@ public class TestNodeController {
}
public static class SecondaryIndexInfo {
- private final int[] primaryKeyIndexes;
- private final PrimaryIndexInfo primaryIndexInfo;
- private final Index secondaryIndex;
- private final ConstantFileSplitProvider fileSplitProvider;
- private final RecordDescriptor rDesc;
- private final int[] insertFieldsPermutations;
+ final int[] primaryKeyIndexes;
+ final PrimaryIndexInfo primaryIndexInfo;
+ final Index secondaryIndex;
+ final ConstantFileSplitProvider fileSplitProvider;
+ final ISerializerDeserializer<?>[] secondaryIndexSerdes;
+ final RecordDescriptor rDesc;
+ final int[] insertFieldsPermutations;
+ final ITypeTraits[] secondaryIndexTypeTraits;
public SecondaryIndexInfo(PrimaryIndexInfo primaryIndexInfo, Index secondaryIndex) {
this.primaryIndexInfo = primaryIndexInfo;
@@ -507,11 +541,11 @@ public class TestNodeController {
FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(),
primaryIndexInfo.dataset, secondaryIndex.getIndexName(), nodes);
fileSplitProvider = new ConstantFileSplitProvider(splits);
- ITypeTraits[] secondaryIndexTypeTraits = createSecondaryIndexTypeTraits(primaryIndexInfo.recordType,
+ secondaryIndexTypeTraits = createSecondaryIndexTypeTraits(primaryIndexInfo.recordType,
primaryIndexInfo.metaType, primaryIndexInfo.primaryKeyTypes,
secondaryIndex.getKeyFieldTypes().toArray(new IAType[secondaryIndex.getKeyFieldTypes().size()]));
- ISerializerDeserializer<?>[] secondaryIndexSerdes = createSecondaryIndexSerdes(primaryIndexInfo.recordType,
- primaryIndexInfo.metaType, primaryIndexInfo.primaryKeyTypes,
+ secondaryIndexSerdes = createSecondaryIndexSerdes(primaryIndexInfo.recordType, primaryIndexInfo.metaType,
+ primaryIndexInfo.primaryKeyTypes,
secondaryIndex.getKeyFieldTypes().toArray(new IAType[secondaryIndex.getKeyFieldTypes().size()]));
rDesc = new RecordDescriptor(secondaryIndexSerdes, secondaryIndexTypeTraits);
insertFieldsPermutations = new int[secondaryIndexTypeTraits.length];
@@ -527,6 +561,10 @@ public class TestNodeController {
public IFileSplitProvider getFileSplitProvider() {
return fileSplitProvider;
}
+
+ public ISerializerDeserializer<?>[] getSerdes() {
+ return secondaryIndexSerdes;
+ }
}
public static class PrimaryIndexInfo {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
index 2eba473..ff027e2 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
@@ -22,7 +22,7 @@ import java.io.DataOutput;
import java.io.IOException;
import java.util.Random;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.om.types.ATypeTag;
public class ABooleanFieldValueGenerator implements IAsterixFieldValueGenerator<Boolean> {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
index e698676..64dab3d 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
@@ -26,7 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.om.types.ATypeTag;
public class ADoubleFieldValueGenerator implements IAsterixFieldValueGenerator<Double> {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
index 7c6556b..5540c11 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
@@ -26,7 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.om.types.ATypeTag;
public class AInt32FieldValueGenerator implements IAsterixFieldValueGenerator<Integer> {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
index 2a2496e..e62054a 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
@@ -26,7 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.om.types.ATypeTag;
public class AInt64FieldValueGenerator implements IAsterixFieldValueGenerator<Long> {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
index cd4de62..b242189 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
@@ -21,7 +21,7 @@ package org.apache.asterix.app.data.gen;
import java.io.DataOutput;
import java.io.IOException;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.builders.RecordBuilder;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
index 5ee6d40..419d1b6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
@@ -26,7 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
-import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.data.gen.RecordTupleGenerator.GenerationFunction;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import org.apache.hyracks.util.string.UTF8StringReader;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/RecordTupleGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/RecordTupleGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/RecordTupleGenerator.java
new file mode 100644
index 0000000..1b04c1a
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/RecordTupleGenerator.java
@@ -0,0 +1,126 @@
+/*
+* 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.app.data.gen;
+
+import java.io.IOException;
+
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestTupleReference;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class RecordTupleGenerator {
+
+ private final int[] keyIndexes;
+ private final int[] keyIndicators;
+ private final ARecordValueGenerator recordGenerator;
+ private final ARecordValueGenerator metaGenerator;
+ private final TestTupleReference tuple;
+
+ public enum GenerationFunction {
+ RANDOM,
+ DETERMINISTIC,
+ INCREASING,
+ DECREASING
+ }
+
+ /**
+ * @param recordType
+ * @param metaType
+ * @param key
+ * @param keyIndexes
+ * @param keyIndicators
+ * @param recordGeneration
+ * @param uniqueRecordFields
+ * @param metaGeneration
+ * @param uniqueMetaFields
+ */
+ public RecordTupleGenerator(ARecordType recordType, ARecordType metaType, int[] keyIndexes, int[] keyIndicators,
+ GenerationFunction[] recordGeneration, boolean[] uniqueRecordFields, GenerationFunction[] metaGeneration,
+ boolean[] uniqueMetaFields) {
+ this.keyIndexes = keyIndexes;
+ this.keyIndicators = keyIndicators;
+ for (IAType field : recordType.getFieldTypes()) {
+ validate(field);
+ }
+ recordGenerator = new ARecordValueGenerator(recordGeneration, recordType, uniqueRecordFields, true);
+ if (metaType != null) {
+ for (IAType field : metaType.getFieldTypes()) {
+ validate(field);
+ }
+ metaGenerator = new ARecordValueGenerator(metaGeneration, metaType, uniqueMetaFields, true);
+ } else {
+ metaGenerator = null;
+ }
+ int numOfFields = keyIndexes.length + 1 + ((metaType != null) ? 1 : 0);
+ tuple = new TestTupleReference(numOfFields);
+ boolean atLeastOneKeyFieldIsNotRandomAndNotBoolean = false;
+ for (int i = 0; i < keyIndexes.length; i++) {
+ if (keyIndicators[i] < 0 || keyIndicators[i] > 1) {
+ throw new IllegalArgumentException("key field indicator must be either 0 or 1");
+ }
+ atLeastOneKeyFieldIsNotRandomAndNotBoolean = atLeastOneKeyFieldIsNotRandomAndNotBoolean
+ || validateKey(keyIndexes[i], keyIndicators[i] == 0 ? recordType : metaType,
+ keyIndicators[i] == 0 ? uniqueRecordFields[i] : uniqueMetaFields[i]);
+ }
+ if (!atLeastOneKeyFieldIsNotRandomAndNotBoolean) {
+ throw new IllegalArgumentException("at least one key field must be unique and not boolean");
+ }
+ if (keyIndexes.length != keyIndicators.length) {
+ throw new IllegalArgumentException("number of key indexes must equals number of key indicators");
+ }
+ }
+
+ private boolean validateKey(int i, ARecordType type, boolean unique) {
+ if (type.getFieldNames().length <= i) {
+ throw new IllegalArgumentException("key index must be less than number of fields");
+ }
+ return unique && type.getFieldTypes()[i].getTypeTag() != ATypeTag.BOOLEAN;
+ }
+
+ public ITupleReference next() throws IOException {
+ tuple.reset();
+ recordGenerator.next(tuple.getFields()[keyIndexes.length].getDataOutput());
+ if (metaGenerator != null) {
+ recordGenerator.next(tuple.getFields()[keyIndexes.length + 1].getDataOutput());
+ }
+ for (int i = 0; i < keyIndexes.length; i++) {
+ if (keyIndicators[i] == 0) {
+ recordGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
+ } else {
+ metaGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
+ }
+ }
+ return tuple;
+ }
+
+ private void validate(IAType field) {
+ switch (field.getTypeTag()) {
+ case BOOLEAN:
+ case DOUBLE:
+ case INTEGER:
+ case BIGINT:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Generating data of type " + field + " is not supported");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
deleted file mode 100644
index 0469349..0000000
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
+++ /dev/null
@@ -1,126 +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.asterix.app.data.gen;
-
-import java.io.IOException;
-
-import org.apache.asterix.om.types.ARecordType;
-import org.apache.asterix.om.types.ATypeTag;
-import org.apache.asterix.om.types.IAType;
-import org.apache.asterix.test.common.TestTupleReference;
-import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-
-public class TupleGenerator {
-
- private final int[] keyIndexes;
- private final int[] keyIndicators;
- private final ARecordValueGenerator recordGenerator;
- private final ARecordValueGenerator metaGenerator;
- private final TestTupleReference tuple;
-
- public enum GenerationFunction {
- RANDOM,
- DETERMINISTIC,
- INCREASING,
- DECREASING
- }
-
- /**
- * @param recordType
- * @param metaType
- * @param key
- * @param keyIndexes
- * @param keyIndicators
- * @param recordGeneration
- * @param uniqueRecordFields
- * @param metaGeneration
- * @param uniqueMetaFields
- */
- public TupleGenerator(ARecordType recordType, ARecordType metaType, int[] keyIndexes, int[] keyIndicators,
- GenerationFunction[] recordGeneration, boolean[] uniqueRecordFields, GenerationFunction[] metaGeneration,
- boolean[] uniqueMetaFields) {
- this.keyIndexes = keyIndexes;
- this.keyIndicators = keyIndicators;
- for (IAType field : recordType.getFieldTypes()) {
- validate(field);
- }
- recordGenerator = new ARecordValueGenerator(recordGeneration, recordType, uniqueRecordFields, true);
- if (metaType != null) {
- for (IAType field : metaType.getFieldTypes()) {
- validate(field);
- }
- metaGenerator = new ARecordValueGenerator(metaGeneration, metaType, uniqueMetaFields, true);
- } else {
- metaGenerator = null;
- }
- int numOfFields = keyIndexes.length + 1 + ((metaType != null) ? 1 : 0);
- tuple = new TestTupleReference(numOfFields);
- boolean atLeastOneKeyFieldIsNotRandomAndNotBoolean = false;
- for (int i = 0; i < keyIndexes.length; i++) {
- if (keyIndicators[i] < 0 || keyIndicators[i] > 1) {
- throw new IllegalArgumentException("key field indicator must be either 0 or 1");
- }
- atLeastOneKeyFieldIsNotRandomAndNotBoolean = atLeastOneKeyFieldIsNotRandomAndNotBoolean
- || validateKey(keyIndexes[i], keyIndicators[i] == 0 ? recordType : metaType,
- keyIndicators[i] == 0 ? uniqueRecordFields[i] : uniqueMetaFields[i]);
- }
- if (!atLeastOneKeyFieldIsNotRandomAndNotBoolean) {
- throw new IllegalArgumentException("at least one key field must be unique and not boolean");
- }
- if (keyIndexes.length != keyIndicators.length) {
- throw new IllegalArgumentException("number of key indexes must equals number of key indicators");
- }
- }
-
- private boolean validateKey(int i, ARecordType type, boolean unique) {
- if (type.getFieldNames().length <= i) {
- throw new IllegalArgumentException("key index must be less than number of fields");
- }
- return unique && type.getFieldTypes()[i].getTypeTag() != ATypeTag.BOOLEAN;
- }
-
- public ITupleReference next() throws IOException {
- tuple.reset();
- recordGenerator.next(tuple.getFields()[keyIndexes.length].getDataOutput());
- if (metaGenerator != null) {
- recordGenerator.next(tuple.getFields()[keyIndexes.length + 1].getDataOutput());
- }
- for (int i = 0; i < keyIndexes.length; i++) {
- if (keyIndicators[i] == 0) {
- recordGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
- } else {
- metaGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
- }
- }
- return tuple;
- }
-
- private void validate(IAType field) {
- switch (field.getTypeTag()) {
- case BOOLEAN:
- case DOUBLE:
- case INTEGER:
- case BIGINT:
- case STRING:
- break;
- default:
- throw new IllegalArgumentException("Generating data of type " + field + " is not supported");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt32ValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt32ValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt32ValueGenerator.java
new file mode 100644
index 0000000..c34f5a3
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt32ValueGenerator.java
@@ -0,0 +1,36 @@
+/*
+ * 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 org.apache.asterix.om.base.AInt32;
+import org.apache.hyracks.storage.am.common.datagen.IFieldValueGenerator;
+
+public class AInt32ValueGenerator implements IFieldValueGenerator<AInt32> {
+ int counter = 0;
+
+ @Override
+ public AInt32 next() {
+ return new AInt32(counter++);
+ }
+
+ @Override
+ public void reset() {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt64ValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt64ValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt64ValueGenerator.java
new file mode 100644
index 0000000..b860737
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/AInt64ValueGenerator.java
@@ -0,0 +1,36 @@
+/*
+ * 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 org.apache.asterix.om.base.AInt64;
+import org.apache.hyracks.storage.am.common.datagen.IFieldValueGenerator;
+
+public class AInt64ValueGenerator implements IFieldValueGenerator<AInt64> {
+ long counter = 0L;
+
+ @Override
+ public AInt64 next() {
+ return new AInt64(counter++);
+ }
+
+ @Override
+ public void reset() {
+ }
+
+}