You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2024/05/11 05:48:19 UTC
(pinot) branch master updated: Make Helix state transition handling idempotent (#12886)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new cd5baa8034 Make Helix state transition handling idempotent (#12886)
cd5baa8034 is described below
commit cd5baa8034b8938431db0d8bd94c979072bce61c
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Fri May 10 22:48:14 2024 -0700
Make Helix state transition handling idempotent (#12886)
---
.../core/data/manager/BaseTableDataManager.java | 562 ++++++++------
.../core/data/manager/InstanceDataManager.java | 32 +-
.../manager/offline/DimensionTableDataManager.java | 43 +-
.../offline/ImmutableSegmentDataManager.java | 5 +
.../manager/offline/OfflineTableDataManager.java | 20 +
.../manager/offline/TableDataManagerProvider.java | 18 +-
.../realtime/RealtimeSegmentDataManager.java | 84 ++-
.../manager/realtime/RealtimeTableDataManager.java | 392 ++++------
.../BaseTableDataManagerAcquireSegmentTest.java | 29 +-
.../data/manager/BaseTableDataManagerTest.java | 809 ++++++++++-----------
.../data/manager/TableDataManagerTestUtils.java | 107 ---
.../offline/DimensionTableDataManagerTest.java | 25 +-
.../realtime/IngestionDelayTrackerTest.java | 31 +-
.../realtime/RealtimeSegmentDataManagerTest.java | 170 ++---
.../realtime/RealtimeTableDataManagerTest.java | 230 ------
...adataAndDictionaryAggregationPlanMakerTest.java | 2 +
.../plan/maker/QueryOverrideWithHintsTest.java | 4 +
.../executor/QueryExecutorExceptionsTest.java | 5 +-
.../core/query/executor/QueryExecutorTest.java | 5 +-
.../pinot/queries/ExplainPlanQueriesTest.java | 5 +-
.../queries/SegmentWithNullValueVectorTest.java | 5 +-
.../testutils/MockInstanceDataManagerFactory.java | 2 +-
.../local/data/manager/SegmentDataManager.java | 17 +
.../local/data/manager/TableDataManager.java | 71 +-
.../indexsegment/immutable/EmptyIndexSegment.java | 4 +
.../immutable/ImmutableSegmentImpl.java | 13 +-
.../immutable/ImmutableSegmentLoader.java | 20 +-
.../indexsegment/mutable/MutableSegmentImpl.java | 11 +-
.../upsert/BasePartitionUpsertMetadataManager.java | 98 +--
.../upsert/PartitionUpsertMetadataManager.java | 2 +
.../pinot/segment/local/utils/SegmentLocks.java | 16 +-
.../org/apache/pinot/segment/spi/IndexSegment.java | 6 +
.../starter/helix/HelixInstanceDataManager.java | 161 ++--
.../helix/SegmentMessageHandlerFactory.java | 4 +-
.../SegmentOnlineOfflineStateModelFactory.java | 188 +----
.../apache/pinot/server/api/BaseResourceTest.java | 6 +-
36 files changed, 1378 insertions(+), 1824 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
index 73af1ed3a0..f9ce484dd4 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
@@ -28,6 +28,7 @@ import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -37,7 +38,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
+import java.util.concurrent.locks.Lock;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.io.FileUtils;
@@ -47,6 +48,7 @@ import org.apache.helix.HelixManager;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.auth.AuthProviderUtils;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMeter;
@@ -62,6 +64,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentContext;
@@ -74,9 +77,9 @@ import org.apache.pinot.segment.spi.store.SegmentDirectory;
import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -86,15 +89,14 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected static final Logger LOGGER = LoggerFactory.getLogger(BaseTableDataManager.class);
protected final ConcurrentHashMap<String, SegmentDataManager> _segmentDataManagerMap = new ConcurrentHashMap<>();
- // Semaphore to restrict the maximum number of parallel segment downloads for a table.
- private Semaphore _segmentDownloadSemaphore;
+ protected final ServerMetrics _serverMetrics = ServerMetrics.get();
protected InstanceDataManagerConfig _instanceDataManagerConfig;
protected String _instanceId;
- protected TableConfig _tableConfig;
protected HelixManager _helixManager;
protected ZkHelixPropertyStore<ZNRecord> _propertyStore;
- protected ServerMetrics _serverMetrics;
+ protected SegmentLocks _segmentLocks;
+ protected TableConfig _tableConfig;
protected String _tableNameWithType;
protected String _tableDataDir;
protected File _indexDir;
@@ -105,6 +107,8 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected String _peerDownloadScheme;
protected long _streamSegmentDownloadUntarRateLimitBytesPerSec;
protected boolean _isStreamSegmentDownloadUntar;
+ // Semaphore to restrict the maximum number of parallel segment downloads for a table
+ private Semaphore _segmentDownloadSemaphore;
// Fixed size LRU cache with TableName - SegmentName pair as key, and segment related
// errors as the value.
@@ -115,17 +119,17 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected volatile boolean _shutDown;
@Override
- public void init(InstanceDataManagerConfig instanceDataManagerConfig, TableConfig tableConfig,
- HelixManager helixManager, @Nullable ExecutorService segmentPreloadExecutor,
+ public void init(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager,
+ SegmentLocks segmentLocks, TableConfig tableConfig, @Nullable ExecutorService segmentPreloadExecutor,
@Nullable LoadingCache<Pair<String, String>, SegmentErrorInfo> errorCache) {
LOGGER.info("Initializing table data manager for table: {}", tableConfig.getTableName());
_instanceDataManagerConfig = instanceDataManagerConfig;
_instanceId = instanceDataManagerConfig.getInstanceId();
_tableConfig = tableConfig;
+ _segmentLocks = segmentLocks;
_helixManager = helixManager;
_propertyStore = helixManager.getHelixPropertyStore();
- _serverMetrics = ServerMetrics.get();
_segmentPreloadExecutor = segmentPreloadExecutor;
_authProvider = AuthProviderUtils.extractAuthProvider(_instanceDataManagerConfig.getAuthConfig(), null);
@@ -182,8 +186,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
doInit();
- _logger.info("Initialized table data manager for table: {} with data directory: {}", _tableNameWithType,
- _tableDataDir);
+ _logger.info("Initialized table data manager with data directory: {}", _tableDataDir);
}
protected abstract void doInit();
@@ -200,9 +203,9 @@ public abstract class BaseTableDataManager implements TableDataManager {
@Override
public synchronized void start() {
- _logger.info("Starting table data manager for table: {}", _tableNameWithType);
+ _logger.info("Starting table data manager");
doStart();
- _logger.info("Started table data manager for table: {}", _tableNameWithType);
+ _logger.info("Started table data manager");
}
protected abstract void doStart();
@@ -210,13 +213,13 @@ public abstract class BaseTableDataManager implements TableDataManager {
@Override
public synchronized void shutDown() {
if (_shutDown) {
- _logger.info("Table data manager for table: {} is already shut down", _tableNameWithType);
+ _logger.warn("Table data manager is already shut down");
return;
}
- _logger.info("Shutting down table data manager for table: {}", _tableNameWithType);
+ _logger.info("Shutting down table data manager");
_shutDown = true;
doShutdown();
- _logger.info("Shut down table data manager for table: {}", _tableNameWithType);
+ _logger.info("Shut down table data manager");
}
protected abstract void doShutdown();
@@ -234,7 +237,10 @@ public abstract class BaseTableDataManager implements TableDataManager {
int numThreads = Math.min(Runtime.getRuntime().availableProcessors(), segmentDataManagers.size());
ExecutorService stopExecutorService = Executors.newFixedThreadPool(numThreads);
for (SegmentDataManager segmentDataManager : segmentDataManagers) {
- stopExecutorService.submit(() -> releaseSegment(segmentDataManager));
+ stopExecutorService.submit(() -> {
+ segmentDataManager.offload();
+ releaseSegment(segmentDataManager);
+ });
}
stopExecutorService.shutdown();
try {
@@ -254,6 +260,16 @@ public abstract class BaseTableDataManager implements TableDataManager {
return _shutDown;
}
+ @Override
+ public Lock getSegmentLock(String segmentName) {
+ return _segmentLocks.getLock(_tableNameWithType, segmentName);
+ }
+
+ @Override
+ public boolean hasSegment(String segmentName) {
+ return _segmentDataManagerMap.containsKey(segmentName);
+ }
+
/**
* {@inheritDoc}
* <p>If one segment already exists with the same name, replaces it with the new one.
@@ -268,7 +284,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
String segmentName = immutableSegment.getSegmentName();
Preconditions.checkState(!_shutDown, "Table data manager is already shut down, cannot add segment: %s to table: %s",
segmentName, _tableNameWithType);
- _logger.info("Adding immutable segment: {} to table: {}", segmentName, _tableNameWithType);
+ _logger.info("Adding immutable segment: {}", segmentName);
_serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.DOCUMENT_COUNT,
immutableSegment.getSegmentMetadata().getTotalDocs());
_serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.SEGMENT_COUNT, 1L);
@@ -276,56 +292,185 @@ public abstract class BaseTableDataManager implements TableDataManager {
ImmutableSegmentDataManager newSegmentManager = new ImmutableSegmentDataManager(immutableSegment);
SegmentDataManager oldSegmentManager = registerSegment(segmentName, newSegmentManager);
if (oldSegmentManager == null) {
- _logger.info("Added new immutable segment: {} to table: {}", segmentName, _tableNameWithType);
+ _logger.info("Added new immutable segment: {}", segmentName);
} else {
- _logger.info("Replaced immutable segment: {} of table: {}", segmentName, _tableNameWithType);
+ _logger.info("Replaced immutable segment: {}", segmentName);
+ oldSegmentManager.offload();
releaseSegment(oldSegmentManager);
}
}
@Override
- public void addSegment(File indexDir, IndexLoadingConfig indexLoadingConfig)
+ public void addOnlineSegment(String segmentName)
throws Exception {
- Preconditions.checkState(!_shutDown, "Table data manager is already shut down, cannot add segment: %s to table: %s",
- indexDir.getName(), _tableNameWithType);
+ Preconditions.checkState(!_shutDown,
+ "Table data manager is already shut down, cannot add ONLINE segment: %s to table: %s", segmentName,
+ _tableNameWithType);
+ _logger.info("Adding ONLINE segment: {}", segmentName);
+ Lock segmentLock = getSegmentLock(segmentName);
+ segmentLock.lock();
+ try {
+ doAddOnlineSegment(segmentName);
+ } catch (Exception e) {
+ addSegmentError(segmentName,
+ new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while adding ONLINE segment", e));
+ throw e;
+ } finally {
+ segmentLock.unlock();
+ }
+ }
+
+ protected abstract void doAddOnlineSegment(String segmentName)
+ throws Exception;
+
+ protected SegmentZKMetadata getZKMetadata(String segmentName) {
+ SegmentZKMetadata zkMetadata =
+ ZKMetadataProvider.getSegmentZKMetadata(_propertyStore, _tableNameWithType, segmentName);
+ Preconditions.checkState(zkMetadata != null, "Failed to find ZK metadata for segment: %s of table: %s", segmentName,
+ _tableNameWithType);
+ return zkMetadata;
+ }
+
+ protected TableConfig getTableConfig() {
+ TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, _tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", _tableNameWithType);
+ return tableConfig;
+ }
+
+ @Nullable
+ protected Schema getSchema(TableConfig tableConfig) {
+ Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, tableConfig);
+ // NOTE: Schema is mandatory for REALTIME table.
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ Preconditions.checkState(schema != null, "Failed to find schema for table: %s", _tableNameWithType);
+ }
+ return schema;
+ }
+
+ protected IndexLoadingConfig getIndexLoadingConfig(@Nullable SegmentZKMetadata zkMetadata) {
+ TableConfig tableConfig = getTableConfig();
+ Schema schema = getSchema(tableConfig);
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
indexLoadingConfig.setTableDataDir(_tableDataDir);
indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
- addSegment(ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, indexLoadingConfig.getSchema()));
+ if (zkMetadata != null) {
+ indexLoadingConfig.setSegmentTier(zkMetadata.getTier());
+ }
+ return indexLoadingConfig;
}
- @Override
- public void addSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, SegmentZKMetadata zkMetadata)
+ /**
+ * Adds a new ONLINE segment that is not already loaded.
+ */
+ protected void addNewOnlineSegment(SegmentZKMetadata zkMetadata, IndexLoadingConfig indexLoadingConfig)
+ throws Exception {
+ String segmentName = zkMetadata.getSegmentName();
+ _logger.info("Adding new ONLINE segment: {}", segmentName);
+ if (!tryLoadExistingSegment(segmentName, indexLoadingConfig, zkMetadata)) {
+ downloadAndLoadSegment(zkMetadata, indexLoadingConfig);
+ }
+ }
+
+ /**
+ * Replaces an already loaded segment in a table if the segment has been overridden in the deep store (CRC mismatch).
+ */
+ protected void replaceSegmentIfCrcMismatch(SegmentDataManager segmentDataManager, SegmentZKMetadata zkMetadata,
+ IndexLoadingConfig indexLoadingConfig)
throws Exception {
- throw new UnsupportedOperationException();
+ String segmentName = segmentDataManager.getSegmentName();
+ Preconditions.checkState(segmentDataManager instanceof ImmutableSegmentDataManager,
+ "Cannot replace CONSUMING segment: %s in table: %s", segmentName, _tableNameWithType);
+ SegmentMetadata localMetadata = segmentDataManager.getSegment().getSegmentMetadata();
+ if (hasSameCRC(zkMetadata, localMetadata)) {
+ _logger.info("Segment: {} has CRC: {} same as before, not replacing it", segmentName, localMetadata.getCrc());
+ return;
+ }
+ _logger.info("Replacing segment: {} because its CRC has changed from: {} to: {}", segmentName,
+ localMetadata.getCrc(), zkMetadata.getCrc());
+ downloadAndLoadSegment(zkMetadata, indexLoadingConfig);
+ _logger.info("Replaced segment: {} with new CRC: {}", segmentName, zkMetadata.getCrc());
}
/**
- * Called when we get a helix transition to go to offline or dropped state.
- * We need to remove it safely, keeping in mind that there may be queries that are
- * using the segment,
- * @param segmentName name of the segment to remove.
+ * Downloads a segment and loads it into the table.
*/
+ protected void downloadAndLoadSegment(SegmentZKMetadata zkMetadata, IndexLoadingConfig indexLoadingConfig)
+ throws Exception {
+ String segmentName = zkMetadata.getSegmentName();
+ _logger.info("Downloading and loading segment: {}", segmentName);
+ File indexDir = downloadSegment(zkMetadata);
+ addSegment(ImmutableSegmentLoader.load(indexDir, indexLoadingConfig));
+ _logger.info("Downloaded and loaded segment: {} with CRC: {} on tier: {}", segmentName, zkMetadata.getCrc(),
+ TierConfigUtils.normalizeTierName(zkMetadata.getTier()));
+ }
+
@Override
- public void removeSegment(String segmentName) {
- // Allow removing segment after shutdown so that we can remove the segment when the table is deleted
+ public void replaceSegment(String segmentName)
+ throws Exception {
+ Preconditions.checkState(!_shutDown,
+ "Table data manager is already shut down, cannot replace segment: %s in table: %s", segmentName,
+ _tableNameWithType);
+ _logger.info("Replacing segment: {}", segmentName);
+ Lock segmentLock = getSegmentLock(segmentName);
+ segmentLock.lock();
+ try {
+ doReplaceSegment(segmentName);
+ } catch (Exception e) {
+ addSegmentError(segmentName,
+ new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while replacing segment", e));
+ throw e;
+ } finally {
+ segmentLock.unlock();
+ }
+ }
+
+ protected void doReplaceSegment(String segmentName)
+ throws Exception {
+ SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
+ if (segmentDataManager != null) {
+ SegmentZKMetadata zkMetadata = getZKMetadata(segmentName);
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(zkMetadata);
+ replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
+ } else {
+ _logger.warn("Failed to find segment: {}, skipping replacing it", segmentName);
+ }
+ }
+
+ @Override
+ public void offloadSegment(String segmentName) {
+ // NOTE: Do not throw exception when data manager has been shut down. This is regular flow when a table is deleted.
if (_shutDown) {
- _logger.info("Table data manager is already shut down, skip removing segment: {} from table: {}", segmentName,
- _tableNameWithType);
+ _logger.info("Table data manager is already shut down, skipping offloading segment: {}", segmentName);
return;
}
- _logger.info("Removing segment: {} from table: {}", segmentName, _tableNameWithType);
+ _logger.info("Offloading segment: {}", segmentName);
+ Lock segmentLock = getSegmentLock(segmentName);
+ segmentLock.lock();
+ try {
+ doOffloadSegment(segmentName);
+ } catch (Exception e) {
+ addSegmentError(segmentName,
+ new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while offloading segment", e));
+ throw e;
+ } finally {
+ segmentLock.unlock();
+ }
+ }
+
+ protected void doOffloadSegment(String segmentName) {
SegmentDataManager segmentDataManager = unregisterSegment(segmentName);
if (segmentDataManager != null) {
+ segmentDataManager.offload();
releaseSegment(segmentDataManager);
- _logger.info("Removed segment: {} from table: {}", segmentName, _tableNameWithType);
+ _logger.info("Offloaded segment: {}", segmentName);
} else {
- _logger.info("Failed to find segment: {} in table: {}", segmentName, _tableNameWithType);
+ _logger.warn("Failed to find segment: {}, skipping offloading it", segmentName);
}
}
/**
* Returns true if the given segment has been deleted recently. The time range is determined by
- * {@link org.apache.pinot.spi.config.instance.InstanceDataManagerConfig#getDeletedSegmentsCacheTtlMinutes()}.
+ * {@link InstanceDataManagerConfig#getDeletedSegmentsCacheTtlMinutes()}.
*/
@Override
public boolean isSegmentDeletedRecently(String segmentName) {
@@ -392,13 +537,13 @@ public abstract class BaseTableDataManager implements TableDataManager {
private void closeSegment(SegmentDataManager segmentDataManager) {
String segmentName = segmentDataManager.getSegmentName();
- _logger.info("Closing segment: {} of table: {}", segmentName, _tableNameWithType);
+ _logger.info("Closing segment: {}", segmentName);
_serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.SEGMENT_COUNT, -1L);
_serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.DELETED_SEGMENT_COUNT, 1L);
_serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.DOCUMENT_COUNT,
-segmentDataManager.getSegment().getSegmentMetadata().getTotalDocs());
segmentDataManager.destroy();
- _logger.info("Closed segment: {} of table: {}", segmentName, _tableNameWithType);
+ _logger.info("Closed segment: {}", segmentName);
}
@Override
@@ -428,17 +573,25 @@ public abstract class BaseTableDataManager implements TableDataManager {
@Override
public void addSegmentError(String segmentName, SegmentErrorInfo segmentErrorInfo) {
- _errorCache.put(Pair.of(_tableNameWithType, segmentName), segmentErrorInfo);
+ if (_errorCache != null) {
+ _errorCache.put(Pair.of(_tableNameWithType, segmentName), segmentErrorInfo);
+ }
}
@Override
public Map<String, SegmentErrorInfo> getSegmentErrors() {
- if (_errorCache == null) {
- return Collections.emptyMap();
+ if (_errorCache != null) {
+ // Filter out entries that match the table name
+ Map<String, SegmentErrorInfo> segmentErrors = new HashMap<>();
+ for (Map.Entry<Pair<String, String>, SegmentErrorInfo> entry : _errorCache.asMap().entrySet()) {
+ Pair<String, String> tableSegmentPair = entry.getKey();
+ if (tableSegmentPair.getLeft().equals(_tableNameWithType)) {
+ segmentErrors.put(tableSegmentPair.getRight(), entry.getValue());
+ }
+ }
+ return segmentErrors;
} else {
- // Filter out entries that match the table name.
- return _errorCache.asMap().entrySet().stream().filter(map -> map.getKey().getLeft().equals(_tableNameWithType))
- .collect(Collectors.toMap(map -> map.getKey().getRight(), Map.Entry::getValue));
+ return Map.of();
}
}
@@ -457,35 +610,37 @@ public abstract class BaseTableDataManager implements TableDataManager {
Preconditions.checkState(!_shutDown,
"Table data manager is already shut down, cannot reload segment: %s of table: %s", segmentName,
_tableNameWithType);
+ _logger.info("Reloading segment: {}", segmentName);
String segmentTier = getSegmentCurrentTier(segmentName);
indexLoadingConfig.setSegmentTier(segmentTier);
indexLoadingConfig.setTableDataDir(_tableDataDir);
indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
File indexDir = getSegmentDataDir(segmentName, segmentTier, indexLoadingConfig.getTableConfig());
+ Lock segmentLock = getSegmentLock(segmentName);
+ segmentLock.lock();
try {
// Download segment from deep store if CRC changes or forced to download;
// otherwise, copy backup directory back to the original index directory.
// And then continue to load the segment from the index directory.
boolean shouldDownload = forceDownload || !hasSameCRC(zkMetadata, localMetadata);
- if (shouldDownload && allowDownload(segmentName, zkMetadata)) {
+ if (shouldDownload) {
// Create backup directory to handle failure of segment reloading.
createBackup(indexDir);
if (forceDownload) {
- LOGGER.info("Segment: {} of table: {} is forced to download", segmentName, _tableNameWithType);
+ _logger.info("Force downloading segment: {}", segmentName);
} else {
- LOGGER.info("Download segment:{} of table: {} as crc changes from: {} to: {}", segmentName,
- _tableNameWithType, localMetadata.getCrc(), zkMetadata.getCrc());
+ _logger.info("Downloading segment: {} because its CRC has changed from: {} to: {}", segmentName,
+ localMetadata.getCrc(), zkMetadata.getCrc());
}
- indexDir = downloadSegment(segmentName, zkMetadata);
+ indexDir = downloadSegment(zkMetadata);
} else {
- LOGGER.info("Reload existing segment: {} of table: {} on tier: {}", segmentName, _tableNameWithType,
+ _logger.info("Reloading existing segment: {} on tier: {}", segmentName,
TierConfigUtils.normalizeTierName(segmentTier));
SegmentDirectory segmentDirectory =
initSegmentDirectory(segmentName, String.valueOf(zkMetadata.getCrc()), indexLoadingConfig);
// We should first try to reuse existing segment directory
if (canReuseExistingDirectoryForReload(zkMetadata, segmentTier, segmentDirectory, indexLoadingConfig, schema)) {
- LOGGER.info("Reloading segment: {} of table: {} using existing segment directory as no reprocessing needed",
- segmentName, _tableNameWithType);
+ _logger.info("Reloading segment: {} using existing segment directory as no reprocessing needed", segmentName);
// No reprocessing needed, reuse the same segment
ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDirectory, indexLoadingConfig, schema);
addSegment(segment);
@@ -506,7 +661,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
// Load from indexDir and replace the old segment in memory. What's inside indexDir
// may come from SegmentDirectory.copyTo() or the segment downloaded from deep store.
indexLoadingConfig.setSegmentTier(zkMetadata.getTier());
- LOGGER.info("Load segment with data from indexDir: {} to tier: {}", indexDir,
+ _logger.info("Loading segment: {} from indexDir: {} to tier: {}", segmentName, indexDir,
TierConfigUtils.normalizeTierName(zkMetadata.getTier()));
ImmutableSegment segment = ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, schema);
addSegment(segment);
@@ -517,11 +672,17 @@ public abstract class BaseTableDataManager implements TableDataManager {
try {
LoaderUtils.reloadFailureRecovery(indexDir);
} catch (Exception recoveryFailureException) {
- LOGGER.error("Failed to recover after reload failure", recoveryFailureException);
+ _logger.error("Failed to recover segment: {} after reload failure", segmentName, recoveryFailureException);
reloadFailureException.addSuppressed(recoveryFailureException);
}
+ addSegmentError(segmentName,
+ new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while reloading segment",
+ reloadFailureException));
throw reloadFailureException;
+ } finally {
+ segmentLock.unlock();
}
+ _logger.info("Reloaded segment: {}", segmentName);
}
private boolean canReuseExistingDirectoryForReload(SegmentZKMetadata segmentZKMetadata, String currentSegmentTier,
@@ -533,51 +694,6 @@ public abstract class BaseTableDataManager implements TableDataManager {
&& !ImmutableSegmentLoader.needPreprocess(segmentDirectory, indexLoadingConfig, schema);
}
- @Override
- public void addOrReplaceSegment(String segmentName, IndexLoadingConfig indexLoadingConfig,
- SegmentZKMetadata zkMetadata, @Nullable SegmentMetadata localMetadata)
- throws Exception {
- Preconditions.checkState(!_shutDown,
- "Table data manager is already shut down, cannot add/replace segment: %s of table: %s", segmentName,
- _tableNameWithType);
- if (localMetadata != null && hasSameCRC(zkMetadata, localMetadata)) {
- LOGGER.info("Segment: {} of table: {} has crc: {} same as before, already loaded, do nothing", segmentName,
- _tableNameWithType, localMetadata.getCrc());
- return;
- }
-
- // The segment is not loaded by the server if the metadata object is null. But the segment
- // may still be kept on the server. For example when server gets restarted, the segment is
- // still on the server but the metadata object has not been initialized yet. In this case,
- // we should check if the segment exists on server and try to load it. If the segment does
- // not exist or fails to get loaded, we download segment from deep store to load it again.
- String segmentTier = zkMetadata.getTier();
- indexLoadingConfig.setSegmentTier(segmentTier);
- indexLoadingConfig.setTableDataDir(_tableDataDir);
- indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
- if (localMetadata == null && tryLoadExistingSegment(segmentName, indexLoadingConfig, zkMetadata)) {
- return;
- }
-
- Preconditions.checkState(allowDownload(segmentName, zkMetadata), "Segment: %s of table: %s does not allow download",
- segmentName, _tableNameWithType);
-
- // Download segment and replace the local one, either due to failure to recover local segment,
- // or the segment data is updated and has new CRC now.
- if (localMetadata == null) {
- LOGGER.info("Download segment: {} of table: {} as it doesn't exist", segmentName, _tableNameWithType);
- } else {
- LOGGER.info("Download segment: {} of table: {} as crc changes from: {} to: {}", segmentName, _tableNameWithType,
- localMetadata.getCrc(), zkMetadata.getCrc());
- }
- File indexDir = downloadSegment(segmentName, zkMetadata);
- ImmutableSegment segment =
- ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, indexLoadingConfig.getSchema(), true);
- addSegment(segment);
- LOGGER.info("Downloaded and loaded segment: {} of table: {} with crc: {} on tier: {}", segmentName,
- _tableNameWithType, zkMetadata.getCrc(), TierConfigUtils.normalizeTierName(segmentTier));
- }
-
/**
* _segmentDataManagerMap is used for fetching segments that need to be queried. If a new segment is created,
* calling this method ensures that all queries in the future can use the new segment. This method may replace an
@@ -608,95 +724,97 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
}
- protected boolean allowDownload(String segmentName, SegmentZKMetadata zkMetadata) {
- return true;
- }
-
- protected File downloadSegment(String segmentName, SegmentZKMetadata zkMetadata)
- throws Exception {
- // TODO: may support download from peer servers for RealTime table.
- return downloadSegmentFromDeepStore(segmentName, zkMetadata);
- }
-
- private File downloadSegmentFromDeepStore(String segmentName, SegmentZKMetadata zkMetadata)
+ /**
+ * Downloads an immutable segment into the index directory.
+ * Segment can be downloaded from deep store or from peer servers. Downloaded segment might be compressed or
+ * encrypted, and this method takes care of decompressing and decrypting the segment.
+ */
+ protected File downloadSegment(SegmentZKMetadata zkMetadata)
throws Exception {
- File tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "-" + UUID.randomUUID());
- if (_isStreamSegmentDownloadUntar && zkMetadata.getCrypterName() == null) {
- try {
- File untaredSegDir = downloadAndStreamUntarWithRateLimit(segmentName, zkMetadata, tempRootDir,
- _streamSegmentDownloadUntarRateLimitBytesPerSec);
- return moveSegment(segmentName, untaredSegDir);
- } finally {
- FileUtils.deleteQuietly(tempRootDir);
- }
- } else {
- try {
- File tarFile = downloadAndDecrypt(segmentName, zkMetadata, tempRootDir);
- return untarAndMoveSegment(segmentName, tarFile, tempRootDir);
- } finally {
- FileUtils.deleteQuietly(tempRootDir);
- }
- }
- }
-
- private File moveSegment(String segmentName, File untaredSegDir)
- throws IOException {
+ String segmentName = zkMetadata.getSegmentName();
+ String downloadUrl = zkMetadata.getDownloadUrl();
+ Preconditions.checkState(downloadUrl != null,
+ "Failed to find download URL in ZK metadata for segment: %s of table: %s", segmentName, _tableNameWithType);
try {
- File indexDir = getSegmentDataDir(segmentName);
- FileUtils.deleteDirectory(indexDir);
- FileUtils.moveDirectory(untaredSegDir, indexDir);
- return indexDir;
+ if (!CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD.equals(downloadUrl)) {
+ try {
+ return downloadSegmentFromDeepStore(zkMetadata);
+ } catch (Exception e) {
+ if (_peerDownloadScheme != null) {
+ _logger.warn("Caught exception while downloading segment: {} from: {}, trying to download from peers",
+ segmentName, downloadUrl, e);
+ return downloadSegmentFromPeers(zkMetadata);
+ } else {
+ throw e;
+ }
+ }
+ } else {
+ return downloadSegmentFromPeers(zkMetadata);
+ }
} catch (Exception e) {
- LOGGER.error("Failed to move segment: {} of table: {}", segmentName, _tableNameWithType);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DIR_MOVEMENT_FAILURES, 1L);
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FAILURES, 1);
throw e;
}
}
- @VisibleForTesting
- File downloadAndDecrypt(String segmentName, SegmentZKMetadata zkMetadata, File tempRootDir)
+ private File downloadSegmentFromDeepStore(SegmentZKMetadata zkMetadata)
throws Exception {
- File tarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
- String uri = zkMetadata.getDownloadUrl();
- boolean downloadSuccess = false;
+ String segmentName = zkMetadata.getSegmentName();
+ String downloadUrl = zkMetadata.getDownloadUrl();
+ _logger.info("Downloading segment: {} from: {}", segmentName, downloadUrl);
+ File tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "-" + UUID.randomUUID());
+ if (_segmentDownloadSemaphore != null) {
+ long startTime = System.currentTimeMillis();
+ _logger.info("Acquiring segment download semaphore for segment: {}, queue-length: {} ", segmentName,
+ _segmentDownloadSemaphore.getQueueLength());
+ _segmentDownloadSemaphore.acquire();
+ _logger.info("Acquired segment download semaphore for segment: {} (lock-time={}ms, queue-length={}).",
+ segmentName, System.currentTimeMillis() - startTime, _segmentDownloadSemaphore.getQueueLength());
+ }
try {
- if (_segmentDownloadSemaphore != null) {
- long startTime = System.currentTimeMillis();
- LOGGER.info("Trying to acquire segment download semaphore for: {}. queue-length: {} ", segmentName,
- _segmentDownloadSemaphore.getQueueLength());
- _segmentDownloadSemaphore.acquire();
- LOGGER.info("Acquired segment download semaphore for: {} (lock-time={}ms, queue-length={}).", segmentName,
- System.currentTimeMillis() - startTime, _segmentDownloadSemaphore.getQueueLength());
- }
- SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(uri, tarFile, zkMetadata.getCrypterName());
- LOGGER.info("Downloaded tarred segment: {} for table: {} from: {} to: {}, file length: {}", segmentName,
- _tableNameWithType, uri, tarFile, tarFile.length());
- downloadSuccess = true;
- return tarFile;
- } catch (AttemptsExceededException e) {
- LOGGER.error("Attempts exceeded when downloading segment: {} for table: {} from: {} to: {}", segmentName,
- _tableNameWithType, uri, tarFile);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES, 1L);
- if (_peerDownloadScheme == null) {
- throw e;
+ File untarredSegmentDir;
+ if (_isStreamSegmentDownloadUntar && zkMetadata.getCrypterName() == null) {
+ _logger.info("Downloading segment: {} using streamed download-untar with maxStreamRateInByte: {}", segmentName,
+ _streamSegmentDownloadUntarRateLimitBytesPerSec);
+ AtomicInteger failedAttempts = new AtomicInteger(0);
+ try {
+ untarredSegmentDir = SegmentFetcherFactory.fetchAndStreamUntarToLocal(downloadUrl, tempRootDir,
+ _streamSegmentDownloadUntarRateLimitBytesPerSec, failedAttempts);
+ _logger.info("Downloaded and untarred segment: {} from: {}, failed attempts: {}", segmentName, downloadUrl,
+ failedAttempts.get());
+ } finally {
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_STREAMED_DOWNLOAD_UNTAR_FAILURES,
+ failedAttempts.get());
+ }
+ } else {
+ File segmentTarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(downloadUrl, segmentTarFile, zkMetadata.getCrypterName());
+ _logger.info("Downloaded tarred segment: {} from: {} to: {}, file length: {}", segmentName, downloadUrl,
+ segmentTarFile, segmentTarFile.length());
+ untarredSegmentDir = untarSegment(segmentName, segmentTarFile, tempRootDir);
}
- downloadFromPeersWithoutStreaming(segmentName, zkMetadata, tarFile);
- downloadSuccess = true;
- return tarFile;
+ File indexDir = moveSegment(segmentName, untarredSegmentDir);
+ _logger.info("Downloaded segment: {} from: {} to: {}", segmentName, downloadUrl, indexDir);
+ return indexDir;
+ } catch (Exception e) {
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES, 1);
+ throw e;
} finally {
- if (!downloadSuccess) {
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FAILURES, 1L);
- }
if (_segmentDownloadSemaphore != null) {
_segmentDownloadSemaphore.release();
}
+ FileUtils.deleteQuietly(tempRootDir);
}
}
- protected void downloadFromPeersWithoutStreaming(String segmentName, SegmentZKMetadata zkMetadata, File destTarFile)
+ private File downloadSegmentFromPeers(SegmentZKMetadata zkMetadata)
throws Exception {
+ String segmentName = zkMetadata.getSegmentName();
Preconditions.checkState(_peerDownloadScheme != null, "Peer download is not enabled for table: %s",
_tableNameWithType);
+ _logger.info("Downloading segment: {} from peers", segmentName);
+ File tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "-" + UUID.randomUUID());
+ File segmentTarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
try {
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(segmentName, _peerDownloadScheme, () -> {
List<URI> peerServerURIs =
@@ -704,76 +822,55 @@ public abstract class BaseTableDataManager implements TableDataManager {
_peerDownloadScheme);
Collections.shuffle(peerServerURIs);
return peerServerURIs;
- }, destTarFile, zkMetadata.getCrypterName());
- _logger.info("Downloaded tarred segment: {} from peers to: {}, file length: {}", segmentName, destTarFile,
- destTarFile.length());
+ }, segmentTarFile, zkMetadata.getCrypterName());
+ _logger.info("Downloaded tarred segment: {} from peers to: {}, file length: {}", segmentName, segmentTarFile,
+ segmentTarFile.length());
+ File indexDir = untarAndMoveSegment(segmentName, segmentTarFile, tempRootDir);
+ _logger.info("Downloaded segment: {} from peers to: {}", segmentName, indexDir);
+ return indexDir;
} catch (Exception e) {
_serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FROM_PEERS_FAILURES, 1);
throw e;
+ } finally {
+ FileUtils.deleteQuietly(tempRootDir);
}
}
- private File downloadAndStreamUntarWithRateLimit(String segmentName, SegmentZKMetadata zkMetadata, File tempRootDir,
- long maxStreamRateInByte)
- throws Exception {
- if (_segmentDownloadSemaphore != null) {
- long startTime = System.currentTimeMillis();
- LOGGER.info("Trying to acquire segment download semaphore for: {}. queue-length: {} ", segmentName,
- _segmentDownloadSemaphore.getQueueLength());
- _segmentDownloadSemaphore.acquire();
- LOGGER.info("Acquired segment download semaphore for: {} (lock-time={}ms, queue-length={}).", segmentName,
- System.currentTimeMillis() - startTime, _segmentDownloadSemaphore.getQueueLength());
- }
- LOGGER.info("Trying to download segment {} using streamed download-untar with maxStreamRateInByte {}", segmentName,
- maxStreamRateInByte);
- String uri = zkMetadata.getDownloadUrl();
- AtomicInteger attempts = new AtomicInteger(0);
+ private File untarSegment(String segmentName, File segmentTarFile, File tempRootDir)
+ throws IOException {
+ File untarDir = new File(tempRootDir, segmentName);
+ _logger.info("Untarring segment: {} from: {} to: {}", segmentName, segmentTarFile, untarDir);
try {
- File ret = SegmentFetcherFactory.fetchAndStreamUntarToLocal(uri, tempRootDir, maxStreamRateInByte, attempts);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_STREAMED_DOWNLOAD_UNTAR_FAILURES,
- attempts.get());
- LOGGER.info("Downloaded and untarred segment: {} for table: {} from: {} attempts: {}", segmentName,
- _tableNameWithType, uri, attempts.get());
- return ret;
+ // If an exception is thrown when untarring, it means the tar file is broken or not found after the retry. Thus,
+ // there's no need to retry again.
+ File untarredSegmentDir = TarGzCompressionUtils.untar(segmentTarFile, untarDir).get(0);
+ _logger.info("Untarred segment: {} into: {}", segmentName, untarredSegmentDir);
+ return untarredSegmentDir;
} catch (Exception e) {
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_STREAMED_DOWNLOAD_UNTAR_FAILURES,
- attempts.get());
- if (e instanceof AttemptsExceededException) {
- LOGGER.error("Attempts exceeded when stream download-untarring segment: {} for table: {} from: {} to: {}",
- segmentName, _tableNameWithType, uri, tempRootDir);
- }
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.UNTAR_FAILURES, 1);
throw e;
- } finally {
- if (_segmentDownloadSemaphore != null) {
- _segmentDownloadSemaphore.release();
- }
}
}
- @VisibleForTesting
- File untarAndMoveSegment(String segmentName, File tarFile, File tempRootDir)
+ private File moveSegment(String segmentName, File untarredSegmentDir)
throws IOException {
- File untarDir = new File(tempRootDir, segmentName);
+ File indexDir = getSegmentDataDir(segmentName);
try {
- // If an exception is thrown when untarring, it means the tar file is broken
- // or not found after the retry. Thus, there's no need to retry again.
- File untaredSegDir = TarGzCompressionUtils.untar(tarFile, untarDir).get(0);
- LOGGER.info("Uncompressed tar file: {} into target dir: {}", tarFile, untarDir);
- // Replace the existing index directory.
- File indexDir = getSegmentDataDir(segmentName);
FileUtils.deleteDirectory(indexDir);
- FileUtils.moveDirectory(untaredSegDir, indexDir);
- LOGGER.info("Successfully downloaded segment: {} of table: {} to index dir: {}", segmentName, _tableNameWithType,
- indexDir);
+ FileUtils.moveDirectory(untarredSegmentDir, indexDir);
return indexDir;
} catch (Exception e) {
- LOGGER.error("Failed to untar segment: {} of table: {} from: {} to: {}", segmentName, _tableNameWithType, tarFile,
- untarDir);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.UNTAR_FAILURES, 1L);
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DIR_MOVEMENT_FAILURES, 1);
throw e;
}
}
+ @VisibleForTesting
+ File untarAndMoveSegment(String segmentName, File segmentTarFile, File tempRootDir)
+ throws IOException {
+ return moveSegment(segmentName, untarSegment(segmentName, segmentTarFile, tempRootDir));
+ }
+
@VisibleForTesting
File getSegmentDataDir(String segmentName) {
return new File(_indexDir, segmentName);
@@ -875,7 +972,6 @@ public abstract class BaseTableDataManager implements TableDataManager {
// Creates the SegmentDirectory object to access the segment metadata.
// The metadata is null if the segment doesn't exist yet.
-
SegmentDirectory segmentDirectory =
tryInitSegmentDirectory(segmentName, String.valueOf(zkMetadata.getCrc()), indexLoadingConfig);
SegmentMetadataImpl segmentMetadata = (segmentDirectory == null) ? null : segmentDirectory.getSegmentMetadata();
@@ -884,10 +980,10 @@ public abstract class BaseTableDataManager implements TableDataManager {
// need to fall back to download the segment from deep store to load it.
if (segmentMetadata == null || !hasSameCRC(zkMetadata, segmentMetadata)) {
if (segmentMetadata == null) {
- LOGGER.info("Segment: {} of table: {} does not exist", segmentName, _tableNameWithType);
+ _logger.info("Segment: {} does not exist", segmentName);
} else if (!hasSameCRC(zkMetadata, segmentMetadata)) {
- LOGGER.info("Segment: {} of table: {} has crc change from: {} to: {}", segmentName, _tableNameWithType,
- segmentMetadata.getCrc(), zkMetadata.getCrc());
+ _logger.info("Segment: {} has CRC changed from: {} to: {}", segmentName, segmentMetadata.getCrc(),
+ zkMetadata.getCrc());
}
closeSegmentDirectoryQuietly(segmentDirectory);
return false;
@@ -899,11 +995,9 @@ public abstract class BaseTableDataManager implements TableDataManager {
// or reprocess it to reflect latest table config and schema before loading.
Schema schema = indexLoadingConfig.getSchema();
if (!ImmutableSegmentLoader.needPreprocess(segmentDirectory, indexLoadingConfig, schema)) {
- LOGGER.info("Segment: {} of table: {} is consistent with latest table config and schema", segmentName,
- _tableNameWithType);
+ _logger.info("Segment: {} is consistent with latest table config and schema", segmentName);
} else {
- LOGGER.info("Segment: {} of table: {} needs reprocess to reflect latest table config and schema", segmentName,
- _tableNameWithType);
+ _logger.info("Segment: {} needs reprocess to reflect latest table config and schema", segmentName);
segmentDirectory.copyTo(indexDir);
// Close the stale SegmentDirectory object and recreate it with reprocessed segment.
closeSegmentDirectoryQuietly(segmentDirectory);
@@ -912,24 +1006,24 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDirectory, indexLoadingConfig, schema);
addSegment(segment);
- LOGGER.info("Loaded existing segment: {} of table: {} with crc: {} on tier: {}", segmentName, _tableNameWithType,
- zkMetadata.getCrc(), TierConfigUtils.normalizeTierName(segmentTier));
+ _logger.info("Loaded existing segment: {} with CRC: {} on tier: {}", segmentName, zkMetadata.getCrc(),
+ TierConfigUtils.normalizeTierName(segmentTier));
return true;
} catch (Exception e) {
- LOGGER.error("Failed to load existing segment: {} of table: {} with crc: {} on tier: {}", segmentName,
- _tableNameWithType, zkMetadata.getCrc(), TierConfigUtils.normalizeTierName(segmentTier), e);
+ _logger.error("Failed to load existing segment: {} with CRC: {} on tier: {}", segmentName, zkMetadata.getCrc(),
+ TierConfigUtils.normalizeTierName(segmentTier), e);
closeSegmentDirectoryQuietly(segmentDirectory);
return false;
}
}
+ @Nullable
private SegmentDirectory tryInitSegmentDirectory(String segmentName, String segmentCrc,
IndexLoadingConfig indexLoadingConfig) {
try {
return initSegmentDirectory(segmentName, segmentCrc, indexLoadingConfig);
} catch (Exception e) {
- LOGGER.warn("Failed to initialize SegmentDirectory for segment: {} of table: {} with error: {}", segmentName,
- _tableNameWithType, e.getMessage());
+ _logger.warn("Failed to initialize SegmentDirectory for segment: {} with error: {}", segmentName, e.getMessage());
return null;
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
index aab7c08b0b..96d8534a65 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
@@ -75,20 +75,37 @@ public interface InstanceDataManager {
throws Exception;
/**
- * Adds a segment into an REALTIME table.
- * <p>The segment might be committed or under consuming.
+ * Adds an ONLINE segment into a table.
+ * This method is triggered by state transition to ONLINE state.
*/
- void addRealtimeSegment(String realtimeTableName, String segmentName)
+ void addOnlineSegment(String tableNameWithType, String segmentName)
+ throws Exception;
+
+ /**
+ * Adds a CONSUMING segment into a REALTIME table.
+ * This method is triggered by state transition to CONSUMING state.
+ */
+ void addConsumingSegment(String realtimeTableName, String segmentName)
+ throws Exception;
+
+ /**
+ * Replaces an already loaded segment in a table if the segment has been overridden in the deep store (CRC mismatch).
+ * This method is triggered by a custom message (NOT state transition), and the target segment should be in ONLINE
+ * state.
+ */
+ void replaceSegment(String tableNameWithType, String segmentName)
throws Exception;
/**
* Offloads a segment from table but not dropping its data from server.
+ * This method is triggered by state transition to OFFLINE state.
*/
void offloadSegment(String tableNameWithType, String segmentName)
throws Exception;
/**
* Delete segment data from the server physically.
+ * This method is triggered by state transition to DROPPED state.
*/
void deleteSegment(String tableNameWithType, String segmentName)
throws Exception;
@@ -118,15 +135,6 @@ public interface InstanceDataManager {
SegmentRefreshSemaphore segmentRefreshSemaphore)
throws Exception;
- /**
- * Adds or replaces a segment in a table. Different from segment reloading, this method
- * doesn't assume the existence of TableDataManager object and it can actually initialize
- * the TableDataManager for the segment. A new segment is downloaded if the local one is
- * not working or has a different CRC from the remote one.
- */
- void addOrReplaceSegment(String tableNameWithType, String segmentName)
- throws Exception;
-
/**
* Returns all tables served by the instance.
*/
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManager.java
index 72b0111c64..2e8fe02c9f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManager.java
@@ -119,41 +119,30 @@ public class DimensionTableDataManager extends OfflineTableDataManager {
@Override
public void addSegment(ImmutableSegment immutableSegment) {
- String segmentName = immutableSegment.getSegmentName();
- Preconditions.checkState(!_shutDown, "Table data manager is already shut down, cannot add segment: %s to table: %s",
- segmentName, _tableNameWithType);
super.addSegment(immutableSegment);
- try {
- if (loadLookupTable()) {
- _logger.info("Successfully loaded lookup table after adding segment: {}", segmentName);
- } else {
- _logger.info("Skip loading lookup table after adding segment: {}, another loading in progress", segmentName);
- }
- } catch (Exception e) {
- throw new RuntimeException(
- String.format("Caught exception while loading lookup table: %s after adding segment: %s", _tableNameWithType,
- segmentName), e);
+ String segmentName = immutableSegment.getSegmentName();
+ if (loadLookupTable()) {
+ _logger.info("Successfully loaded lookup table after adding segment: {}", segmentName);
+ } else {
+ _logger.info("Skip loading lookup table after adding segment: {}, another loading in progress", segmentName);
}
}
@Override
- public void removeSegment(String segmentName) {
- // Allow removing segment after shutdown so that we can remove the segment when the table is deleted
- if (_shutDown) {
- _logger.info("Table data manager is already shut down, skip removing segment: {}", segmentName);
- return;
- }
- super.removeSegment(segmentName);
- try {
+ protected void doOffloadSegment(String segmentName) {
+ SegmentDataManager segmentDataManager = unregisterSegment(segmentName);
+ if (segmentDataManager != null) {
+ segmentDataManager.offload();
+ releaseSegment(segmentDataManager);
+ _logger.info("Offloaded segment: {}", segmentName);
if (loadLookupTable()) {
- _logger.info("Successfully loaded lookup table after removing segment: {}", segmentName);
+ _logger.info("Successfully loaded lookup table after offloading segment: {}", segmentName);
} else {
- _logger.info("Skip loading lookup table after removing segment: {}, another loading in progress", segmentName);
+ _logger.info("Skip loading lookup table after offloading segment: {}, another loading in progress",
+ segmentName);
}
- } catch (Exception e) {
- throw new RuntimeException(
- String.format("Caught exception while loading lookup table: %s after removing segment: %s",
- _tableNameWithType, segmentName), e);
+ } else {
+ _logger.warn("Failed to find segment: {}, skipping offloading it", segmentName);
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/ImmutableSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/ImmutableSegmentDataManager.java
index 1646ba77b9..6dbb66630c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/ImmutableSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/ImmutableSegmentDataManager.java
@@ -43,6 +43,11 @@ public class ImmutableSegmentDataManager extends SegmentDataManager {
return _immutableSegment;
}
+ @Override
+ public void doOffload() {
+ _immutableSegment.offload();
+ }
+
@Override
protected void doDestroy() {
_immutableSegment.destroy();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java
index aef0d80b9b..6123f82f5c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java
@@ -19,7 +19,10 @@
package org.apache.pinot.core.data.manager.offline;
import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.core.data.manager.BaseTableDataManager;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
/**
@@ -40,4 +43,21 @@ public class OfflineTableDataManager extends BaseTableDataManager {
protected void doShutdown() {
releaseAndRemoveAllSegments();
}
+
+ protected void doAddOnlineSegment(String segmentName)
+ throws Exception {
+ SegmentZKMetadata zkMetadata = getZKMetadata(segmentName);
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(zkMetadata);
+ SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
+ if (segmentDataManager == null) {
+ addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+ } else {
+ replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
+ }
+ }
+
+ @Override
+ public void addConsumingSegment(String segmentName) {
+ throw new UnsupportedOperationException("Cannot add CONSUMING segment to OFFLINE table");
+ }
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/TableDataManagerProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/TableDataManagerProvider.java
index f3cfa41ee7..e2b9a190de 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/TableDataManagerProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/TableDataManagerProvider.java
@@ -30,6 +30,7 @@ import org.apache.helix.HelixManager;
import org.apache.pinot.common.restlet.resources.SegmentErrorInfo;
import org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
@@ -42,20 +43,24 @@ import org.apache.pinot.spi.utils.IngestionConfigUtils;
*/
public class TableDataManagerProvider {
private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final HelixManager _helixManager;
+ private final SegmentLocks _segmentLocks;
private final Semaphore _segmentBuildSemaphore;
- public TableDataManagerProvider(InstanceDataManagerConfig instanceDataManagerConfig) {
+ public TableDataManagerProvider(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager,
+ SegmentLocks segmentLocks) {
_instanceDataManagerConfig = instanceDataManagerConfig;
+ _helixManager = helixManager;
+ _segmentLocks = segmentLocks;
int maxParallelSegmentBuilds = instanceDataManagerConfig.getMaxParallelSegmentBuilds();
_segmentBuildSemaphore = maxParallelSegmentBuilds > 0 ? new Semaphore(maxParallelSegmentBuilds, true) : null;
}
- public TableDataManager getTableDataManager(TableConfig tableConfig, HelixManager helixManager) {
- return getTableDataManager(tableConfig, helixManager, null, null, () -> true);
+ public TableDataManager getTableDataManager(TableConfig tableConfig) {
+ return getTableDataManager(tableConfig, null, null, () -> true);
}
- public TableDataManager getTableDataManager(TableConfig tableConfig, HelixManager helixManager,
- @Nullable ExecutorService segmentPreloadExecutor,
+ public TableDataManager getTableDataManager(TableConfig tableConfig, @Nullable ExecutorService segmentPreloadExecutor,
@Nullable LoadingCache<Pair<String, String>, SegmentErrorInfo> errorCache,
Supplier<Boolean> isServerReadyToServeQueries) {
TableDataManager tableDataManager;
@@ -80,7 +85,8 @@ public class TableDataManagerProvider {
default:
throw new IllegalStateException();
}
- tableDataManager.init(_instanceDataManagerConfig, tableConfig, helixManager, segmentPreloadExecutor, errorCache);
+ tableDataManager.init(_instanceDataManagerConfig, _helixManager, _segmentLocks, tableConfig, segmentPreloadExecutor,
+ errorCache);
return tableDataManager;
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index 3290c5e4f3..e68e595ec8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -34,6 +34,7 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
import java.util.function.BooleanSupplier;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
@@ -722,7 +723,6 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
_state = State.HOLDING;
SegmentCompletionProtocol.Response response = postSegmentConsumedMsg();
SegmentCompletionProtocol.ControllerResponseStatus status = response.getStatus();
- boolean success;
switch (status) {
case NOT_LEADER:
// Retain the same state
@@ -757,14 +757,20 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
_state = State.DISCARDED;
break;
case DEFAULT:
- success = buildSegmentAndReplace();
- if (success) {
- _state = State.RETAINED;
- } else {
- // Could not build segment for some reason. We can only download it.
- _state = State.ERROR;
- _segmentLogger.error("Could not build segment for {}", _segmentNameStr);
+ // Lock the segment to avoid multiple threads touching the same segment.
+ Lock segmentLock = _realtimeTableDataManager.getSegmentLock(_segmentNameStr);
+ segmentLock.lock();
+ try {
+ if (buildSegmentAndReplace()) {
+ _state = State.RETAINED;
+ break;
+ }
+ } finally {
+ segmentLock.unlock();
}
+ // Could not build segment for some reason. We can only download it.
+ _state = State.ERROR;
+ _segmentLogger.error("Could not build segment for {}", _segmentNameStr);
break;
default:
break;
@@ -773,24 +779,30 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
case COMMIT:
_state = State.COMMITTING;
_currentOffset = _partitionGroupConsumer.checkpoint(_currentOffset);
- long buildTimeSeconds = response.getBuildTimeSeconds();
- buildSegmentForCommit(buildTimeSeconds * 1000L);
- if (_segmentBuildDescriptor == null) {
- // We could not build the segment. Go into error state.
- _state = State.ERROR;
- _segmentLogger.error("Could not build segment for {}", _segmentNameStr);
- } else {
- success = commitSegment(response.getControllerVipUrl());
- if (success) {
+ // Lock the segment to avoid multiple threads touching the same segment.
+ Lock segmentLock = _realtimeTableDataManager.getSegmentLock(_segmentNameStr);
+ segmentLock.lock();
+ try {
+ long buildTimeSeconds = response.getBuildTimeSeconds();
+ buildSegmentForCommit(buildTimeSeconds * 1000L);
+ if (_segmentBuildDescriptor == null) {
+ // We could not build the segment. Go into error state.
+ _state = State.ERROR;
+ _segmentLogger.error("Could not build segment for {}", _segmentNameStr);
+ break;
+ }
+ if (commitSegment(response.getControllerVipUrl())) {
_state = State.COMMITTED;
- } else {
- // If for any reason commit failed, we don't want to be in COMMITTING state when we hold.
- // Change the state to HOLDING before looping around.
- _state = State.HOLDING;
- _segmentLogger.info("Could not commit segment. Retrying after hold");
- hold();
+ break;
}
+ } finally {
+ segmentLock.unlock();
}
+ // If for any reason commit failed, we don't want to be in COMMITTING state when we hold.
+ // Change the state to HOLDING before looping around.
+ _state = State.HOLDING;
+ _segmentLogger.info("Could not commit segment. Retrying after hold");
+ hold();
break;
default:
_segmentLogger.error("Holding after response from Controller: {}", response.toJsonString());
@@ -1066,7 +1078,8 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
}
@VisibleForTesting
- boolean commitSegment(String controllerVipUrl) {
+ boolean commitSegment(String controllerVipUrl)
+ throws Exception {
File segmentTarFile = _segmentBuildDescriptor.getSegmentTarFile();
Preconditions.checkState(segmentTarFile != null && segmentTarFile.exists(), "Segment tar file: %s does not exist",
segmentTarFile);
@@ -1076,7 +1089,7 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_SUCCESS);
return false;
}
- _realtimeTableDataManager.replaceLLSegment(_segmentNameStr, _indexLoadingConfig);
+ _realtimeTableDataManager.replaceConsumingSegment(_segmentNameStr);
removeSegmentFile();
return true;
}
@@ -1104,12 +1117,13 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
return segmentCommitter.commit(_segmentBuildDescriptor);
}
- protected boolean buildSegmentAndReplace() {
+ protected boolean buildSegmentAndReplace()
+ throws Exception {
SegmentBuildDescriptor descriptor = buildSegmentInternal(false);
if (descriptor == null) {
return false;
}
- _realtimeTableDataManager.replaceLLSegment(_segmentNameStr, _indexLoadingConfig);
+ _realtimeTableDataManager.replaceConsumingSegment(_segmentNameStr);
return true;
}
@@ -1298,10 +1312,10 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
}
}
- protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) {
+ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
+ throws Exception {
closeStreamConsumers();
- _realtimeTableDataManager
- .downloadAndReplaceSegment(_segmentNameStr, segmentZKMetadata, _indexLoadingConfig, _tableConfig);
+ _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata);
}
protected long now() {
@@ -1331,15 +1345,21 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
return true;
}
- protected void doDestroy() {
+ @Override
+ public void doOffload() {
try {
stop();
} catch (InterruptedException e) {
_segmentLogger.error("Could not stop consumer thread");
}
- _realtimeSegment.destroy();
closeStreamConsumers();
cleanupMetrics();
+ _realtimeSegment.offload();
+ }
+
+ @Override
+ protected void doDestroy() {
+ _realtimeSegment.destroy();
}
public void startConsumption() {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
index b120867d6b..6997295aaa 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
@@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.io.File;
import java.io.IOException;
-import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -44,33 +43,29 @@ import org.apache.pinot.common.Utils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.restlet.resources.SegmentErrorInfo;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.SegmentUtils;
-import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
-import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
import org.apache.pinot.core.data.manager.BaseTableDataManager;
import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager;
-import org.apache.pinot.core.util.PeerServerSegmentFinder;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.dedup.PartitionDedupMetadataManager;
import org.apache.pinot.segment.local.dedup.TableDedupMetadataManager;
import org.apache.pinot.segment.local.dedup.TableDedupMetadataManagerFactory;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
import org.apache.pinot.segment.local.segment.virtualcolumn.VirtualColumnProviderFactory;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import org.apache.pinot.segment.local.upsert.TableUpsertMetadataManager;
import org.apache.pinot.segment.local.upsert.TableUpsertMetadataManagerFactory;
import org.apache.pinot.segment.local.utils.SchemaUtils;
-import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.local.utils.tablestate.TableStateUtils;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentContext;
-import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
import org.apache.pinot.spi.config.table.IndexingConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -83,8 +78,6 @@ import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.spi.utils.TimeUtils;
-import static org.apache.pinot.spi.utils.CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD;
-
@ThreadSafe
public class RealtimeTableDataManager extends BaseTableDataManager {
@@ -119,7 +112,6 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
private static final int MIN_INTERVAL_BETWEEN_STATS_UPDATES_MINUTES = 30;
public static final long READY_TO_CONSUME_DATA_CHECK_INTERVAL_MS = TimeUnit.SECONDS.toMillis(5);
- private static final SegmentLocks SEGMENT_UPSERT_LOCKS = SegmentLocks.create();
// TODO: Change it to BooleanSupplier
private final Supplier<Boolean> _isServerReadyToServeQueries;
@@ -150,8 +142,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
try {
_statsHistory = RealtimeSegmentStatsHistory.deserialzeFrom(statsFile);
} catch (IOException | ClassNotFoundException e) {
- _logger.error("Error reading history object for table {} from {}", _tableNameWithType,
- statsFile.getAbsolutePath(), e);
+ _logger.error("Caught exception while reading stats history from: {}", statsFile.getAbsolutePath(), e);
File savedFile = new File(_tableDataDir, STATS_FILE_NAME + "." + UUID.randomUUID());
try {
FileUtils.moveFile(statsFile, savedFile);
@@ -258,7 +249,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
try {
_tableUpsertMetadataManager.close();
} catch (IOException e) {
- _logger.warn("Cannot close upsert metadata manager properly for table: {}", _tableNameWithType, e);
+ _logger.warn("Caught exception while closing upsert metadata manager", e);
}
} else {
releaseAndRemoveAllSegments();
@@ -377,95 +368,98 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
&& upsertConfig.isEnableSnapshot() && upsertConfig.isEnablePreload();
}
- /*
- * This call comes in one of two ways:
- * For HL Segments:
- * - We are being directed by helix to own up all the segments that we committed and are still in retention. In
- * this case we treat it exactly like how OfflineTableDataManager would -- wrap it into an
- * OfflineSegmentDataManager, and put it in the map.
- * - We are being asked to own up a new realtime segment. In this case, we wrap the segment with a
- * RealTimeSegmentDataManager (that kicks off consumption). When the segment is committed we get notified via the
- * notifySegmentCommitted call, at which time we replace the segment with the OfflineSegmentDataManager
- *
- * For LL Segments:
- * - We are being asked to start consuming from a partition.
- * - We did not know about the segment and are being asked to download and own the segment (re-balancing, or
- * replacing a realtime server with a fresh one, maybe). We need to look at segment metadata and decide whether
- * to start consuming or download the segment.
+ /**
+ * Handles upsert preload, and returns whether the upsert preload is enabled.
*/
- @Override
- public void addSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, SegmentZKMetadata segmentZKMetadata)
- throws Exception {
- Preconditions.checkState(!_shutDown, "Table data manager is already shut down, cannot add segment: %s to table: %s",
- segmentName, _tableNameWithType);
- boolean upsertPreloadEnabled = isUpsertPreloadEnabled();
- if (upsertPreloadEnabled) {
- Integer partitionId = SegmentUtils.getRealtimeSegmentPartitionId(segmentName, segmentZKMetadata, null);
- Preconditions.checkNotNull(partitionId,
- String.format("Failed to get partition id for segment: %s (upsert-enabled table: %s)", segmentName,
- _tableNameWithType));
- PartitionUpsertMetadataManager partitionUpsertMetadataManager =
- _tableUpsertMetadataManager.getOrCreatePartitionManager(partitionId);
- partitionUpsertMetadataManager.preloadSegments(indexLoadingConfig);
- // Continue to add segment after preloading, as the segment might not be added by preloading.
+ private void handleUpsertPreload(SegmentZKMetadata zkMetadata, IndexLoadingConfig indexLoadingConfig) {
+ if (!isUpsertPreloadEnabled()) {
+ return;
}
+ String segmentName = zkMetadata.getSegmentName();
+ Integer partitionId = SegmentUtils.getRealtimeSegmentPartitionId(segmentName, zkMetadata, null);
+ Preconditions.checkState(partitionId != null,
+ String.format("Failed to get partition id for segment: %s in upsert-enabled table: %s", segmentName,
+ _tableNameWithType));
+ _tableUpsertMetadataManager.getOrCreatePartitionManager(partitionId).preloadSegments(indexLoadingConfig);
+ }
+
+ protected void doAddOnlineSegment(String segmentName)
+ throws Exception {
+ SegmentZKMetadata zkMetadata = getZKMetadata(segmentName);
+ Preconditions.checkState(zkMetadata.getStatus() != Status.IN_PROGRESS,
+ "Segment: %s of table: %s is not committed, cannot make it ONLINE", segmentName, _tableNameWithType);
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(zkMetadata);
+ handleUpsertPreload(zkMetadata, indexLoadingConfig);
SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
- if (segmentDataManager != null) {
- if (upsertPreloadEnabled) {
- _logger.debug(
- "Skipping adding existing segment: {} for table: {} with data manager class: {}, as it's preloaded",
- segmentName, _tableNameWithType, segmentDataManager.getClass().getSimpleName());
+ if (segmentDataManager == null) {
+ addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+ } else {
+ if (segmentDataManager instanceof RealtimeSegmentDataManager) {
+ _logger.info("Changing segment: {} from CONSUMING to ONLINE", segmentName);
+ ((RealtimeSegmentDataManager) segmentDataManager).goOnlineFromConsuming(zkMetadata);
+ onConsumingToOnline(segmentName);
} else {
- _logger.warn("Skipping adding existing segment: {} for table: {} with data manager class: {}", segmentName,
- _tableNameWithType, segmentDataManager.getClass().getSimpleName());
+ replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
}
+ }
+ }
+
+ @Override
+ public void addConsumingSegment(String segmentName) {
+ Preconditions.checkState(!_shutDown,
+ "Table data manager is already shut down, cannot add CONSUMING segment: %s to table: %s", segmentName,
+ _tableNameWithType);
+ _logger.info("Adding CONSUMING segment: {}", segmentName);
+ Lock segmentLock = getSegmentLock(segmentName);
+ segmentLock.lock();
+ try {
+ doAddConsumingSegment(segmentName);
+ } catch (Exception e) {
+ addSegmentError(segmentName,
+ new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while adding CONSUMING segment", e));
+ throw e;
+ } finally {
+ segmentLock.unlock();
+ }
+ }
+
+ private void doAddConsumingSegment(String segmentName) {
+ SegmentZKMetadata zkMetadata = getZKMetadata(segmentName);
+ if (zkMetadata.getStatus() != Status.IN_PROGRESS) {
+ // NOTE: We do not throw exception here because the segment might have just been committed before the state
+ // transition is processed. We can skip adding this segment, and the segment will enter CONSUMING state in
+ // Helix, then we can rely on the following CONSUMING -> ONLINE state transition to add it.
+ _logger.warn("Segment: {} is already committed, skipping adding it as CONSUMING segment", segmentName);
+ return;
+ }
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(zkMetadata);
+ handleUpsertPreload(zkMetadata, indexLoadingConfig);
+ SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
+ if (segmentDataManager != null) {
+ _logger.warn("Segment: {} ({}) already exists, skipping adding it as CONSUMING segment", segmentName,
+ segmentDataManager instanceof RealtimeSegmentDataManager ? "CONSUMING" : "COMPLETED");
return;
}
- // Assign table directory and tier info to not let the segment be moved during loading/preprocessing
- indexLoadingConfig.setTableDataDir(_tableDataDir);
- indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
- indexLoadingConfig.setSegmentTier(segmentZKMetadata.getTier());
+ _logger.info("Adding new CONSUMING segment: {}", segmentName);
+ // Delete the segment directory if it already exists
File segmentDir = new File(_indexDir, segmentName);
- // Restart during segment reload might leave segment in inconsistent state (index directory might not exist but
- // segment backup directory existed), need to first try to recover from reload failure before checking the existence
- // of the index directory and loading segment from it
- LoaderUtils.reloadFailureRecovery(segmentDir);
+ FileUtils.deleteQuietly(segmentDir);
TableConfig tableConfig = indexLoadingConfig.getTableConfig();
Schema schema = indexLoadingConfig.getSchema();
- assert schema != null;
- if (segmentZKMetadata.getStatus().isCompleted()) {
- if (tryLoadExistingSegment(segmentName, indexLoadingConfig, segmentZKMetadata)) {
- // The existing completed segment has been loaded successfully
- return;
- } else {
- // For LLC and uploaded segments, delete the local copy and download a new copy
- _logger.info("Unable to load local LLC segment: {}, downloading a new copy", segmentName);
- FileUtils.deleteQuietly(segmentDir);
- }
- // Local segment doesn't exist or cannot load, download a new copy
- downloadAndReplaceSegment(segmentName, segmentZKMetadata, indexLoadingConfig, tableConfig);
- return;
- } else {
- // Metadata has not been committed, delete the local segment if exists
- FileUtils.deleteQuietly(segmentDir);
- }
-
- // Start a new consuming segment
- if (!isValid(schema, tableConfig.getIndexingConfig())) {
- _logger.error("Not adding segment {}", segmentName);
- throw new RuntimeException("Mismatching schema/table config for " + _tableNameWithType);
- }
-
+ assert tableConfig != null && schema != null;
+ validate(tableConfig, schema);
VirtualColumnProviderFactory.addBuiltInVirtualColumnsToSegmentSchema(schema, segmentName);
- setDefaultTimeValueIfInvalid(tableConfig, schema, segmentZKMetadata);
+ setDefaultTimeValueIfInvalid(tableConfig, schema, zkMetadata);
- // Generates only one semaphore for every partitionGroupId
+ // Generates only one semaphore for every partition
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
int partitionGroupId = llcSegmentName.getPartitionGroupId();
Semaphore semaphore = _partitionGroupIdToSemaphoreMap.computeIfAbsent(partitionGroupId, k -> new Semaphore(1));
+
+ // Create the segment data manager and register it
PartitionUpsertMetadataManager partitionUpsertMetadataManager =
_tableUpsertMetadataManager != null ? _tableUpsertMetadataManager.getOrCreatePartitionManager(partitionGroupId)
: null;
@@ -473,15 +467,14 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
_tableDedupMetadataManager != null ? _tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId)
: null;
RealtimeSegmentDataManager realtimeSegmentDataManager =
- new RealtimeSegmentDataManager(segmentZKMetadata, tableConfig, this, _indexDir.getAbsolutePath(),
- indexLoadingConfig, schema, llcSegmentName, semaphore, _serverMetrics, partitionUpsertMetadataManager,
+ new RealtimeSegmentDataManager(zkMetadata, tableConfig, this, _indexDir.getAbsolutePath(), indexLoadingConfig,
+ schema, llcSegmentName, semaphore, _serverMetrics, partitionUpsertMetadataManager,
partitionDedupMetadataManager, _isTableReadyToConsumeData);
realtimeSegmentDataManager.startConsumption();
- segmentDataManager = realtimeSegmentDataManager;
+ _serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.SEGMENT_COUNT, 1);
+ registerSegment(segmentName, realtimeSegmentDataManager);
- _logger.info("Initialized RealtimeSegmentDataManager - " + segmentName);
- registerSegment(segmentName, segmentDataManager);
- _serverMetrics.addValueToTableGauge(_tableNameWithType, ServerGauge.SEGMENT_COUNT, 1L);
+ _logger.info("Added new CONSUMING segment: {}", segmentName);
}
/**
@@ -548,7 +541,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
private void handleUpsert(ImmutableSegment immutableSegment) {
String segmentName = immutableSegment.getSegmentName();
- _logger.info("Adding immutable segment: {} to upsert-enabled table: {}", segmentName, _tableNameWithType);
+ _logger.info("Adding immutable segment: {} with upsert enabled", segmentName);
Integer partitionId =
SegmentUtils.getRealtimeSegmentPartitionId(segmentName, _tableNameWithType, _helixManager, null);
@@ -567,168 +560,61 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
// Besides, preloading happens before the table partition is made ready for any queries.
partitionUpsertMetadataManager.preloadSegment(immutableSegment);
registerSegment(segmentName, newSegmentManager);
- _logger.info("Preloaded immutable segment: {} to upsert-enabled table: {}", segmentName, _tableNameWithType);
+ _logger.info("Preloaded immutable segment: {} with upsert enabled", segmentName);
return;
}
- // Replacing segment takes multiple steps, and particularly need to access the oldSegment. Replace segment may
- // happen in two threads, i.e. the consuming thread that's committing the mutable segment and a HelixTaskExecutor
- // thread that's bringing segment from ONLINE to CONSUMING when the server finds consuming thread can't commit
- // the segment in time. The slower thread takes the reference of the oldSegment here, but it may get closed by
- // the faster thread if not synchronized. In particular, the slower thread may iterate the primary keys in the
- // oldSegment, causing seg fault. So we have to take a lock here.
- // However, we can't just reuse the existing segmentLocks. Because many methods of partitionUpsertMetadataManager
- // takes this lock internally, but after taking snapshot RW lock. If we take segmentLock here (before taking
- // snapshot RW lock), we can get into deadlock with threads calling partitionUpsertMetadataManager's other
- // methods, like removeSegment.
- // Adding segment should be done by a single HelixTaskExecutor thread, but do it with lock here for simplicity
- // otherwise, we'd need to double-check if oldSegmentManager is null.
- Lock segmentLock = SEGMENT_UPSERT_LOCKS.getLock(_tableNameWithType, segmentName);
- segmentLock.lock();
- try {
- SegmentDataManager oldSegmentManager = _segmentDataManagerMap.get(segmentName);
- if (oldSegmentManager == null) {
- // When adding a new segment, we should register it 'before' it is fully initialized by
- // partitionUpsertMetadataManager. Because when processing docs in the new segment, the docs in the other
- // segments may be invalidated, making the queries see less valid docs than expected. We should let query
- // access the new segment asap even though its validDocId bitmap is still being filled by
- // partitionUpsertMetadataManager.
- registerSegment(segmentName, newSegmentManager);
- partitionUpsertMetadataManager.addSegment(immutableSegment);
- _logger.info("Added new immutable segment: {} to upsert-enabled table: {}", segmentName, _tableNameWithType);
- } else {
- // When replacing a segment, we should register the new segment 'after' it is fully initialized by
- // partitionUpsertMetadataManager to fill up its validDocId bitmap. Otherwise, the queries will lose the access
- // to the valid docs in the old segment immediately, but the validDocId bitmap of the new segment is still
- // being filled by partitionUpsertMetadataManager, making the queries see less valid docs than expected.
- // When replacing a segment, the new and old segments are assumed to have same set of valid docs for data
- // consistency, otherwise the new segment should be named differently to go through the addSegment flow above.
- IndexSegment oldSegment = oldSegmentManager.getSegment();
- partitionUpsertMetadataManager.replaceSegment(immutableSegment, oldSegment);
- registerSegment(segmentName, newSegmentManager);
- _logger.info("Replaced {} segment: {} of upsert-enabled table: {}",
- oldSegment instanceof ImmutableSegment ? "immutable" : "mutable", segmentName, _tableNameWithType);
- releaseSegment(oldSegmentManager);
- }
- } finally {
- segmentLock.unlock();
- }
- }
-
- @Override
- protected boolean allowDownload(String segmentName, SegmentZKMetadata zkMetadata) {
- // Cannot download consuming segment
- if (zkMetadata.getStatus() == Status.IN_PROGRESS) {
- return false;
- }
- // TODO: may support download from peer servers as well.
- return !METADATA_URI_FOR_PEER_DOWNLOAD.equals(zkMetadata.getDownloadUrl());
- }
-
- void downloadAndReplaceSegment(String segmentName, SegmentZKMetadata segmentZKMetadata,
- IndexLoadingConfig indexLoadingConfig, TableConfig tableConfig) {
- String uri = segmentZKMetadata.getDownloadUrl();
- if (!METADATA_URI_FOR_PEER_DOWNLOAD.equals(uri)) {
- try {
- // TODO: cleanup and consolidate the segment loading logic a bit for OFFLINE and REALTIME tables.
- // https://github.com/apache/pinot/issues/9752
- downloadSegmentFromDeepStore(segmentName, indexLoadingConfig, uri);
- } catch (Exception e) {
- _logger.warn("Download segment {} from deepstore uri {} failed.", segmentName, uri, e);
- // Download from deep store failed; try to download from peer if peer download is setup for the table.
- if (_peerDownloadScheme != null) {
- downloadSegmentFromPeer(segmentName, indexLoadingConfig);
- } else {
- throw e;
- }
- }
+ SegmentDataManager oldSegmentManager = _segmentDataManagerMap.get(segmentName);
+ if (oldSegmentManager == null) {
+ // When adding a new segment, we should register it 'before' it is fully initialized by
+ // partitionUpsertMetadataManager. Because when processing docs in the new segment, the docs in the other
+ // segments may be invalidated, making the queries see less valid docs than expected. We should let query
+ // access the new segment asap even though its validDocId bitmap is still being filled by
+ // partitionUpsertMetadataManager.
+ registerSegment(segmentName, newSegmentManager);
+ partitionUpsertMetadataManager.addSegment(immutableSegment);
+ _logger.info("Added new immutable segment: {} with upsert enabled", segmentName);
} else {
- if (_peerDownloadScheme != null) {
- downloadSegmentFromPeer(segmentName, indexLoadingConfig);
- } else {
- throw new RuntimeException("Peer segment download not enabled for segment " + segmentName);
- }
- }
- }
-
- private void downloadSegmentFromDeepStore(String segmentName, IndexLoadingConfig indexLoadingConfig, String uri) {
- // This could leave temporary directories in _indexDir if JVM shuts down before the temp directory is deleted.
- // This is fine since the temporary directories are deleted when the table data manager calls init.
- File tempRootDir = null;
- try {
- tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "." + System.currentTimeMillis());
- File segmentTarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
- SegmentFetcherFactory.fetchSegmentToLocal(uri, segmentTarFile);
- _logger.info("Downloaded file from {} to {}; Length of downloaded file: {}", uri, segmentTarFile,
- segmentTarFile.length());
- untarAndMoveSegment(segmentName, indexLoadingConfig, segmentTarFile, tempRootDir);
- } catch (Exception e) {
- _logger.warn("Failed to download segment {} from deep store: ", segmentName, e);
- throw new RuntimeException(e);
- } finally {
- FileUtils.deleteQuietly(tempRootDir);
+ // When replacing a segment, we should register the new segment 'after' it is fully initialized by
+ // partitionUpsertMetadataManager to fill up its validDocId bitmap. Otherwise, the queries will lose the access
+ // to the valid docs in the old segment immediately, but the validDocId bitmap of the new segment is still
+ // being filled by partitionUpsertMetadataManager, making the queries see less valid docs than expected.
+ // When replacing a segment, the new and old segments are assumed to have same set of valid docs for data
+ // consistency, otherwise the new segment should be named differently to go through the addSegment flow above.
+ IndexSegment oldSegment = oldSegmentManager.getSegment();
+ partitionUpsertMetadataManager.replaceSegment(immutableSegment, oldSegment);
+ registerSegment(segmentName, newSegmentManager);
+ _logger.info("Replaced {} segment: {} with upsert enabled",
+ oldSegment instanceof ImmutableSegment ? "immutable" : "mutable", segmentName);
+ oldSegmentManager.offload();
+ releaseSegment(oldSegmentManager);
}
}
/**
- * Untars the new segment and replaces the existing segment.
+ * Replaces the CONSUMING segment with a downloaded committed one.
*/
- private void untarAndMoveSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, File segmentTarFile,
- File tempRootDir)
- throws IOException {
- File untarDir = new File(tempRootDir, segmentName);
- File untaredSegDir = TarGzCompressionUtils.untar(segmentTarFile, untarDir).get(0);
- _logger.info("Uncompressed file {} into tmp dir {}", segmentTarFile, untarDir);
- File indexDir = new File(_indexDir, segmentName);
- FileUtils.deleteQuietly(indexDir);
- FileUtils.moveDirectory(untaredSegDir, indexDir);
- _logger.info("Replacing LLC Segment {}", segmentName);
- replaceLLSegment(segmentName, indexLoadingConfig);
- }
-
- private void downloadSegmentFromPeer(String segmentName, IndexLoadingConfig indexLoadingConfig) {
- File tempRootDir = null;
- try {
- tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "." + System.currentTimeMillis());
- File segmentTarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
- // Next download the segment from a randomly chosen server using configured download scheme (http or https).
- SegmentFetcherFactory.getSegmentFetcher(_peerDownloadScheme).fetchSegmentToLocal(segmentName, () -> {
- List<URI> peerServerURIs =
- PeerServerSegmentFinder.getPeerServerURIs(_helixManager, _tableNameWithType, segmentName,
- _peerDownloadScheme);
- Collections.shuffle(peerServerURIs);
- return peerServerURIs;
- }, segmentTarFile);
- _logger.info("Fetched segment {} successfully to {} of size {}", segmentName, segmentTarFile,
- segmentTarFile.length());
- untarAndMoveSegment(segmentName, indexLoadingConfig, segmentTarFile, tempRootDir);
- } catch (Exception e) {
- _logger.warn("Download and move segment {} from peer with scheme {} failed.", segmentName, _peerDownloadScheme,
- e);
- throw new RuntimeException(e);
- } finally {
- FileUtils.deleteQuietly(tempRootDir);
- }
+ public void downloadAndReplaceConsumingSegment(SegmentZKMetadata zkMetadata)
+ throws Exception {
+ String segmentName = zkMetadata.getSegmentName();
+ _logger.info("Downloading and replacing CONSUMING segment: {} with committed one", segmentName);
+ File indexDir = downloadSegment(zkMetadata);
+ // Get a new index loading config with latest table config and schema to load the segment
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(zkMetadata);
+ addSegment(ImmutableSegmentLoader.load(indexDir, indexLoadingConfig));
+ _logger.info("Downloaded and replaced CONSUMING segment: {}", segmentName);
}
/**
- * Replaces a committed LLC REALTIME segment.
+ * Replaces the CONSUMING segment with the one sealed locally.
*/
- public void replaceLLSegment(String segmentName, IndexLoadingConfig indexLoadingConfig) {
+ public void replaceConsumingSegment(String segmentName)
+ throws Exception {
+ _logger.info("Replacing CONSUMING segment: {} with the one sealed locally", segmentName);
File indexDir = new File(_indexDir, segmentName);
- // Use the latest table config and schema to load the segment
- TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, _tableNameWithType);
- Preconditions.checkState(tableConfig != null, "Failed to get table config for table: %s", _tableNameWithType);
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, tableConfig);
-
- // Construct a new indexLoadingConfig with the updated tableConfig and schema.
- InstanceDataManagerConfig instanceDataManagerConfig = indexLoadingConfig.getInstanceDataManagerConfig();
- IndexLoadingConfig newIndexLoadingConfig = new IndexLoadingConfig(instanceDataManagerConfig, tableConfig, schema);
-
- try {
- addSegment(indexDir, newIndexLoadingConfig);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ // Get a new index loading config with latest table config and schema to load the segment
+ IndexLoadingConfig indexLoadingConfig = getIndexLoadingConfig(null);
+ addSegment(ImmutableSegmentLoader.load(indexDir, indexLoadingConfig));
+ _logger.info("Replaced CONSUMING segment: {}", segmentName);
}
public String getServerInstance() {
@@ -767,31 +653,23 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
*
* If we add more validations, it may make sense to split this method into multiple validation methods.
* But then, we are trying to figure out all the invalid cases before we return from this method...
- *
- * @return true if schema is valid.
*/
- private boolean isValid(Schema schema, IndexingConfig indexingConfig) {
+ private void validate(TableConfig tableConfig, Schema schema) {
// 1. Make sure that the sorted column is not a multi-value field.
+ IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
List<String> sortedColumns = indexingConfig.getSortedColumn();
- boolean isValid = true;
if (CollectionUtils.isNotEmpty(sortedColumns)) {
- final String sortedColumn = sortedColumns.get(0);
+ String sortedColumn = sortedColumns.get(0);
if (sortedColumns.size() > 1) {
_logger.warn("More than one sorted column configured. Using {}", sortedColumn);
}
FieldSpec fieldSpec = schema.getFieldSpecFor(sortedColumn);
- if (!fieldSpec.isSingleValueField()) {
- _logger.error("Cannot configure multi-valued column {} as sorted column", sortedColumn);
- isValid = false;
- }
- }
- // 2. We want to get the schema errors, if any, even if isValid is false;
- try {
- SchemaUtils.validate(schema);
- } catch (Exception e) {
- _logger.error("Caught exception while validating schema: {}", schema.getSchemaName(), e);
- isValid = false;
+ Preconditions.checkArgument(fieldSpec != null, "Failed to find sorted column: %s in schema for table: %s",
+ sortedColumn, _tableNameWithType);
+ Preconditions.checkArgument(fieldSpec.isSingleValueField(),
+ "Cannot configure multi-valued column %s as sorted column for table: %s", sortedColumn, _tableNameWithType);
}
- return isValid;
+ // 2. Validate the schema itself
+ SchemaUtils.validate(schema);
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerAcquireSegmentTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerAcquireSegmentTest.java
index a245c740be..50546f57b1 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerAcquireSegmentTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerAcquireSegmentTest.java
@@ -22,14 +22,12 @@ import com.google.common.collect.ImmutableList;
import java.io.File;
import java.lang.reflect.Field;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.metrics.ServerMetrics;
@@ -37,6 +35,7 @@ import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager;
import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
@@ -66,12 +65,9 @@ public class BaseTableDataManagerAcquireSegmentTest {
// Set once for every test
private volatile int _nDestroys;
private volatile boolean _closing;
- private Set<ImmutableSegment> _allSegments = new HashSet<>();
- private Set<SegmentDataManager> _accessedSegManagers =
- Collections.newSetFromMap(new ConcurrentHashMap<SegmentDataManager, Boolean>());
- private Set<SegmentDataManager> _allSegManagers =
- Collections.newSetFromMap(new ConcurrentHashMap<SegmentDataManager, Boolean>());
- private AtomicInteger _numQueries = new AtomicInteger(0);
+ private final Set<ImmutableSegment> _allSegments = new HashSet<>();
+ private final Set<SegmentDataManager> _accessedSegManagers = ConcurrentHashMap.newKeySet();
+ private final Set<SegmentDataManager> _allSegManagers = ConcurrentHashMap.newKeySet();
private Map<String, ImmutableSegmentDataManager> _internalSegMap;
private Throwable _exception;
private Thread _masterThread;
@@ -110,7 +106,6 @@ public class BaseTableDataManagerAcquireSegmentTest {
_allSegments.clear();
_accessedSegManagers.clear();
_allSegManagers.clear();
- _numQueries.set(0);
_exception = null;
_masterThread = null;
}
@@ -123,7 +118,8 @@ public class BaseTableDataManagerAcquireSegmentTest {
when(instanceDataManagerConfig.getDeletedSegmentsCacheTtlMinutes()).thenReturn(DELETED_SEGMENTS_TTL_MINUTES);
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
TableDataManager tableDataManager = new OfflineTableDataManager();
- tableDataManager.init(instanceDataManagerConfig, tableConfig, mock(HelixManager.class), null, null);
+ tableDataManager.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), tableConfig, null,
+ null);
tableDataManager.start();
Field segsMapField = BaseTableDataManager.class.getDeclaredField("_segmentDataManagerMap");
segsMapField.setAccessible(true);
@@ -159,7 +155,7 @@ public class BaseTableDataManagerAcquireSegmentTest {
Assert.assertEquals(tableDataManager.getNumSegments(), 1);
SegmentDataManager segmentDataManager = tableDataManager.acquireSegment(segmentName);
Assert.assertEquals(segmentDataManager.getReferenceCount(), 2);
- tableDataManager.removeSegment(segmentName);
+ tableDataManager.offloadSegment(segmentName);
Assert.assertEquals(tableDataManager.getNumSegments(), 0);
Assert.assertEquals(segmentDataManager.getReferenceCount(), 1);
Assert.assertEquals(_nDestroys, 0);
@@ -184,10 +180,10 @@ public class BaseTableDataManagerAcquireSegmentTest {
// return false.
tableDataManager.addSegment(immutableSegment);
Assert.assertFalse(tableDataManager.isSegmentDeletedRecently(segmentName));
- tableDataManager.removeSegment(segmentName);
+ tableDataManager.offloadSegment(segmentName);
// Removing the segment again is fine.
- tableDataManager.removeSegment(segmentName);
+ tableDataManager.offloadSegment(segmentName);
Assert.assertEquals(tableDataManager.getNumSegments(), 0);
// Add a new segment and remove it in order this time.
@@ -219,7 +215,7 @@ public class BaseTableDataManagerAcquireSegmentTest {
// Delete ix2 without accessing it.
SegmentDataManager sdm2 = _internalSegMap.get(anotherSeg);
Assert.assertEquals(sdm2.getReferenceCount(), 1);
- tableDataManager.removeSegment(anotherSeg);
+ tableDataManager.offloadSegment(anotherSeg);
Assert.assertEquals(tableDataManager.getNumSegments(), 0);
Assert.assertEquals(sdm2.getReferenceCount(), 0);
verify(ix2, times(1)).destroy();
@@ -453,10 +449,11 @@ public class BaseTableDataManagerAcquireSegmentTest {
}
// Remove the segment _lo and then bump _lo
- private void removeSegment() {
+ private void removeSegment()
+ throws Exception {
// Keep at least one segment in place.
if (_hi > _lo) {
- _tableDataManager.removeSegment(SEGMENT_PREFIX + _lo);
+ _tableDataManager.offloadSegment(SEGMENT_PREFIX + _lo);
_lo++;
} else {
addSegment();
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
index ace744c999..1282e493fd 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
@@ -18,13 +18,13 @@
*/
package org.apache.pinot.core.data.manager;
+import java.io.DataInputStream;
import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -37,52 +37,68 @@ import org.apache.pinot.common.tier.TierFactory;
import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.common.utils.fetcher.BaseSegmentFetcher;
import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager;
import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
+import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TierConfig;
import org.apache.pinot.spi.crypt.PinotCrypter;
import org.apache.pinot.spi.crypt.PinotCrypterFactory;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
import org.apache.pinot.util.TestUtils;
-import org.mockito.MockedStatic;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import static org.mockito.Mockito.*;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.*;
public class BaseTableDataManagerTest {
private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "BaseTableDataManagerTest");
- private static final String TABLE_NAME = "table01";
- private static final String TABLE_NAME_WITH_TYPE = "table01_OFFLINE";
- private static final File TABLE_DATA_DIR = new File(TEMP_DIR, TABLE_NAME_WITH_TYPE);
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String OFFLINE_TABLE_NAME = TableNameBuilder.OFFLINE.tableNameWithType(RAW_TABLE_NAME);
+ private static final File TABLE_DATA_DIR = new File(TEMP_DIR, OFFLINE_TABLE_NAME);
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final String TIER_SEGMENT_DIRECTORY_LOADER = "tierBased";
+ private static final String TIER_NAME = "coolTier";
private static final String STRING_COLUMN = "col1";
private static final String[] STRING_VALUES = {"A", "D", "E", "B", "C"};
private static final String LONG_COLUMN = "col2";
private static final long[] LONG_VALUES = {10000L, 20000L, 50000L, 40000L, 30000L};
+ private static final TableConfig DEFAULT_TABLE_CONFIG =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+ private static final TableConfig TIER_TABLE_CONFIG =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setTierConfigList(List.of(
+ new TierConfig(TIER_NAME, TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "3d", null,
+ TierFactory.PINOT_SERVER_STORAGE_TYPE, "tag_OFFLINE", null,
+ Map.of("dataDir", new File(TEMP_DIR, TIER_NAME).getAbsolutePath())))).build();
+ private static final Schema SCHEMA =
+ new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME).addSingleValueDimension(STRING_COLUMN, DataType.STRING)
+ .addMetric(LONG_COLUMN, DataType.LONG).build();
+
@BeforeClass
public void setUp()
throws Exception {
@@ -93,7 +109,7 @@ public class BaseTableDataManagerTest {
public void setUpMethod()
throws Exception {
TestUtils.ensureDirectoriesExistAndEmpty(TEMP_DIR);
- TableDataManagerTestUtils.initSegmentFetcher();
+ initSegmentFetcher();
}
@AfterMethod
@@ -102,88 +118,88 @@ public class BaseTableDataManagerTest {
FileUtils.deleteDirectory(TEMP_DIR);
}
+ public static void initSegmentFetcher()
+ throws Exception {
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(BaseSegmentFetcher.RETRY_COUNT_CONFIG_KEY, 3);
+ properties.put(BaseSegmentFetcher.RETRY_WAIT_MS_CONFIG_KEY, 100);
+ properties.put(BaseSegmentFetcher.RETRY_DELAY_SCALE_FACTOR_CONFIG_KEY, 5);
+ SegmentFetcherFactory.init(new PinotConfiguration(properties));
+
+ // Setup crypter
+ properties.put("class.fakePinotCrypter", BaseTableDataManagerTest.FakePinotCrypter.class.getName());
+ PinotCrypterFactory.init(new PinotConfiguration(properties));
+ }
+
@Test
public void testReloadSegmentNewData()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- SegmentZKMetadata zkmd = createRawSegment(tableConfig, segName, SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
// Mock the case where segment is loaded but its CRC is different from
// the one in zk, thus raw segment is downloaded and loaded.
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn("0");
- BaseTableDataManager tmgr = createTableManager();
- assertFalse(tmgr.getSegmentDataDir(segName).exists());
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd, null, false);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File dataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME);
+ assertFalse(dataDir.exists());
+ tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, localMetadata, null, false);
+ assertTrue(dataDir.exists());
+ assertEquals(new SegmentMetadataImpl(dataDir).getTotalDocs(), 5);
}
@Test
public void testReloadSegmentNewDataNewTier()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- SegmentZKMetadata zkmd = createRawSegment(tableConfig, segName, SegmentVersion.v3, 5);
- String tierName = "coolTier";
- zkmd.setTier(tierName);
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
+ zkMetadata.setTier(TIER_NAME);
// Mock the case where segment is loaded but its CRC is different from
// the one in zk, thus raw segment is downloaded and loaded.
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn("0");
// No dataDir for coolTier, thus stay on default tier.
- BaseTableDataManager tmgr = createTableManager();
- File defaultSegDir = tmgr.getSegmentDataDir(segName);
- assertFalse(defaultSegDir.exists());
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null),
- zkmd, llmd, null, false);
- assertTrue(defaultSegDir.exists());
- llmd = new SegmentMetadataImpl(defaultSegDir);
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File defaultDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME);
+ assertFalse(defaultDataDir.exists());
+ tableDataManager.reloadSegment(SEGMENT_NAME, createTierIndexLoadingConfig(DEFAULT_TABLE_CONFIG), zkMetadata,
+ localMetadata, null, false);
+ assertTrue(defaultDataDir.exists());
+ assertEquals(new SegmentMetadataImpl(defaultDataDir).getTotalDocs(), 5);
// Configured dataDir for coolTier, thus move to new dir.
- llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
- tableConfig = createTableConfigWithTier(tierName, new File(TEMP_DIR, tierName));
- tmgr = createTableManager();
- IndexLoadingConfig loadingCfg = TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null);
- tmgr.reloadSegment(segName, loadingCfg, zkmd, llmd, null, false);
- File segDirOnTier = tmgr.getSegmentDataDir(segName, tierName, loadingCfg.getTableConfig());
- assertTrue(segDirOnTier.exists());
- assertFalse(defaultSegDir.exists());
- llmd = new SegmentMetadataImpl(segDirOnTier);
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), segDirOnTier);
+ tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, createTierIndexLoadingConfig(TIER_TABLE_CONFIG), zkMetadata,
+ localMetadata, null, false);
+ File tierDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME, TIER_NAME, TIER_TABLE_CONFIG);
+ assertTrue(tierDataDir.exists());
+ assertFalse(defaultDataDir.exists());
+ assertEquals(new SegmentMetadataImpl(tierDataDir).getTotalDocs(), 5);
}
@Test
public void testReloadSegmentUseLocalCopy()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v1, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v1);
+ File indexDir = createSegment(SegmentVersion.v1, 5);
+ long crc = getCRC(indexDir);
// Same CRCs so load the local segment directory directly.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn(Long.toString(segCrc));
-
- BaseTableDataManager tmgr = createTableManager();
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd, null, false);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
-
- FileUtils.deleteQuietly(localSegDir);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn(Long.toString(crc));
+
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, localMetadata, null, false);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
+
+ FileUtils.deleteQuietly(indexDir);
try {
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd, null, false);
+ tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, localMetadata, null, false);
fail();
} catch (Exception e) {
// As expected, segment reloading fails due to missing the local segment dir.
@@ -193,227 +209,200 @@ public class BaseTableDataManagerTest {
@Test
public void testReloadSegmentUseLocalCopyNewTier()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v1, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v1);
+ File indexDir = createSegment(SegmentVersion.v1, 5);
+ long crc = getCRC(indexDir);
// Same CRCs so load the local segment directory directly.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- String tierName = "coolTier";
- when(zkmd.getTier()).thenReturn(tierName);
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn(Long.toString(segCrc));
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ when(zkMetadata.getTier()).thenReturn(TIER_NAME);
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn(Long.toString(crc));
// No dataDir for coolTier, thus stay on default tier.
- BaseTableDataManager tmgr = createTableManager();
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null),
- zkmd, llmd, null, false);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), localSegDir);
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, createTierIndexLoadingConfig(DEFAULT_TABLE_CONFIG), zkMetadata,
+ localMetadata, null, false);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
// Configured dataDir for coolTier, thus move to new dir.
- llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn(Long.toString(segCrc));
- tableConfig = createTableConfigWithTier(tierName, new File(TEMP_DIR, tierName));
- tmgr = createTableManager();
- IndexLoadingConfig loadingCfg = TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null);
- tmgr.reloadSegment(segName, loadingCfg, zkmd, llmd, null, false);
- File segDirOnTier = tmgr.getSegmentDataDir(segName, tierName, loadingCfg.getTableConfig());
- assertTrue(segDirOnTier.exists());
- assertFalse(localSegDir.exists());
- llmd = new SegmentMetadataImpl(segDirOnTier);
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), segDirOnTier);
+ tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, createTierIndexLoadingConfig(TIER_TABLE_CONFIG), zkMetadata,
+ localMetadata, null, false);
+ File tierDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME, TIER_NAME, TIER_TABLE_CONFIG);
+ assertTrue(tierDataDir.exists());
+ assertFalse(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(tierDataDir).getTotalDocs(), 5);
}
@Test
public void testReloadSegmentConvertVersion()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v1, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v1);
+ File indexDir = createSegment(SegmentVersion.v1, 5);
+ long crc = getCRC(indexDir);
// Same CRCs so load the local segment directory directly.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn(Long.toString(segCrc));
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn(Long.toString(crc));
// Require to use v3 format.
- IndexLoadingConfig idxCfg = TableDataManagerTestUtils.createIndexLoadingConfig();
- idxCfg.setSegmentVersion(SegmentVersion.v3);
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
+ indexLoadingConfig.setSegmentVersion(SegmentVersion.v3);
- BaseTableDataManager tmgr = createTableManager();
- tmgr.reloadSegment(segName, idxCfg, zkmd, llmd, null, false);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getVersion(), SegmentVersion.v3);
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, indexLoadingConfig, zkMetadata, localMetadata, null, false);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ assertEquals(segmentMetadata.getVersion(), SegmentVersion.v3);
}
@Test
public void testReloadSegmentAddIndex()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
- assertFalse(hasInvertedIndex(localSegDir, STRING_COLUMN, SegmentVersion.v3));
- assertFalse(hasInvertedIndex(localSegDir, LONG_COLUMN, SegmentVersion.v3));
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
+ assertFalse(hasInvertedIndex(indexDir, STRING_COLUMN));
+ assertFalse(hasInvertedIndex(indexDir, LONG_COLUMN));
// Same CRCs so load the local segment directory directly.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn(Long.toString(segCrc));
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ SegmentMetadata localMetadata = mock(SegmentMetadata.class);
+ when(localMetadata.getCrc()).thenReturn(Long.toString(crc));
// Require to add indices.
- IndexLoadingConfig idxCfg = TableDataManagerTestUtils.createIndexLoadingConfig();
- idxCfg.setSegmentVersion(SegmentVersion.v3);
- idxCfg.setInvertedIndexColumns(new HashSet<>(Arrays.asList(STRING_COLUMN, LONG_COLUMN)));
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
+ indexLoadingConfig.setInvertedIndexColumns(new HashSet<>(Arrays.asList(STRING_COLUMN, LONG_COLUMN)));
- BaseTableDataManager tmgr = createTableManager();
- tmgr.reloadSegment(segName, idxCfg, zkmd, llmd, null, false);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
- assertTrue(hasInvertedIndex(tmgr.getSegmentDataDir(segName), STRING_COLUMN, SegmentVersion.v3));
- assertTrue(hasInvertedIndex(tmgr.getSegmentDataDir(segName), LONG_COLUMN, SegmentVersion.v3));
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.reloadSegment(SEGMENT_NAME, indexLoadingConfig, zkMetadata, localMetadata, null, false);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
+ assertTrue(hasInvertedIndex(indexDir, STRING_COLUMN));
+ assertTrue(hasInvertedIndex(indexDir, LONG_COLUMN));
}
@Test
public void testReloadSegmentForceDownload()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- SegmentZKMetadata zkmd = TableDataManagerTestUtils.makeRawSegment(segName, localSegDir,
- new File(TEMP_DIR, segName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION), false);
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata =
+ makeRawSegment(indexDir, new File(TEMP_DIR, SEGMENT_NAME + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION), false);
// Same CRC but force to download.
- BaseTableDataManager tmgr = createTableManager();
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getCrc(), zkmd.getCrc() + "");
+ BaseTableDataManager tableDataManager = createTableManager();
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(Long.parseLong(segmentMetadata.getCrc()), zkMetadata.getCrc());
// Remove the local segment dir. Segment reloading fails unless force to download.
- FileUtils.deleteQuietly(localSegDir);
+ FileUtils.deleteQuietly(indexDir);
try {
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd, null, false);
+ tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, segmentMetadata, null, false);
fail();
} catch (Exception e) {
// As expected, segment reloading fails due to missing the local segment dir.
}
- tmgr.reloadSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd, null, true);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
-
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getCrc(), zkmd.getCrc() + "");
- assertEquals(llmd.getTotalDocs(), 5);
+ tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, segmentMetadata, null, true);
+ assertTrue(indexDir.exists());
+ segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(Long.parseLong(segmentMetadata.getCrc()), zkMetadata.getCrc());
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
}
@Test
- public void testAddOrReplaceSegmentNewData()
+ public void testReplaceSegmentNewData()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- SegmentZKMetadata zkmd = createRawSegment(tableConfig, segName, SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
// Mock the case where segment is loaded but its CRC is different from
// the one in zk, thus raw segment is downloaded and loaded.
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
+ ImmutableSegmentDataManager segmentDataManager = createImmutableSegmentDataManager(SEGMENT_NAME, 0);
- BaseTableDataManager tmgr = createTableManager();
- assertFalse(tmgr.getSegmentDataDir(segName).exists());
- tmgr.addOrReplaceSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File dataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME);
+ assertFalse(dataDir.exists());
+ tableDataManager.replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, new IndexLoadingConfig());
+ assertTrue(dataDir.exists());
+ assertEquals(new SegmentMetadataImpl(dataDir).getTotalDocs(), 5);
}
@Test
- public void testAddOrReplaceSegmentNewDataNewTier()
+ public void testReplaceSegmentNewDataNewTier()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- SegmentZKMetadata zkmd = createRawSegment(tableConfig, segName, SegmentVersion.v3, 5);
- String tierName = "coolTier";
- zkmd.setTier(tierName);
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
+ zkMetadata.setTier(TIER_NAME);
// Mock the case where segment is loaded but its CRC is different from
// the one in zk, thus raw segment is downloaded and loaded.
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
+ ImmutableSegmentDataManager segmentDataManager = createImmutableSegmentDataManager(SEGMENT_NAME, 0);
// No dataDir for coolTier, thus stay on default tier.
- BaseTableDataManager tmgr = createTableManager();
- File defaultSegDir = tmgr.getSegmentDataDir(segName);
- assertFalse(defaultSegDir.exists());
- tmgr.addOrReplaceSegment(segName,
- TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null), zkmd, llmd);
- assertTrue(defaultSegDir.exists());
- llmd = new SegmentMetadataImpl(defaultSegDir);
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File defaultDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME);
+ assertFalse(defaultDataDir.exists());
+ tableDataManager.replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata,
+ createTierIndexLoadingConfig(DEFAULT_TABLE_CONFIG));
+ assertTrue(defaultDataDir.exists());
+ assertEquals(new SegmentMetadataImpl(defaultDataDir).getTotalDocs(), 5);
// Configured dataDir for coolTier, thus move to new dir.
- llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("0");
- tableConfig = createTableConfigWithTier(tierName, new File(TEMP_DIR, tierName));
- tmgr = createTableManager();
- IndexLoadingConfig loadingCfg = TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null);
- tmgr.addOrReplaceSegment(segName, loadingCfg, zkmd, llmd);
- File segDirOnTier = tmgr.getSegmentDataDir(segName, tierName, loadingCfg.getTableConfig());
- assertTrue(segDirOnTier.exists());
- assertFalse(defaultSegDir.exists());
- llmd = new SegmentMetadataImpl(segDirOnTier);
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), segDirOnTier);
+ tableDataManager = createTableManager();
+ tableDataManager.replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata,
+ createTierIndexLoadingConfig(TIER_TABLE_CONFIG));
+ File tierDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME, TIER_NAME, TIER_TABLE_CONFIG);
+ assertTrue(tierDataDir.exists());
+ assertFalse(defaultDataDir.exists());
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(tierDataDir);
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ assertEquals(segmentMetadata.getIndexDir(), tierDataDir);
}
@Test
- public void testAddOrReplaceSegmentNoop()
+ public void testReplaceSegmentNoop()
throws Exception {
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(Long.valueOf(1024));
+ String segmentName = "seg01";
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(segmentName);
+ when(zkMetadata.getCrc()).thenReturn(1024L);
- SegmentMetadata llmd = mock(SegmentMetadata.class);
- when(llmd.getCrc()).thenReturn("1024");
+ ImmutableSegmentDataManager segmentDataManager = createImmutableSegmentDataManager(segmentName, 1024L);
- BaseTableDataManager tmgr = createTableManager();
- assertFalse(tmgr.getSegmentDataDir("seg01").exists());
- tmgr.addOrReplaceSegment("seg01", TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, llmd);
+ BaseTableDataManager tableDataManager = createTableManager();
+ assertFalse(tableDataManager.getSegmentDataDir(segmentName).exists());
+ tableDataManager.replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, new IndexLoadingConfig());
// As CRC is same, the index dir is left as is, so not get created by the test.
- assertFalse(tmgr.getSegmentDataDir("seg01").exists());
+ assertFalse(tableDataManager.getSegmentDataDir(segmentName).exists());
}
@Test
- public void testAddOrReplaceSegmentUseLocalCopy()
+ public void testAddNewSegmentUseLocalCopy()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- when(zkmd.getDownloadUrl()).thenReturn("file://somewhere");
-
- BaseTableDataManager tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
-
- FileUtils.deleteQuietly(localSegDir);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ when(zkMetadata.getDownloadUrl()).thenReturn("file://somewhere");
+
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, new IndexLoadingConfig());
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
+
+ FileUtils.deleteQuietly(indexDir);
try {
- tmgr.addOrReplaceSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, null);
+ tableDataManager.addNewOnlineSegment(zkMetadata, new IndexLoadingConfig());
fail();
} catch (Exception e) {
// As expected, when local segment dir is missing, it tries to download
@@ -423,249 +412,214 @@ public class BaseTableDataManagerTest {
}
@Test
- public void testAddOrReplaceSegmentUseLocalCopyNewTier()
+ public void testAddSegmentUseLocalCopyNewTier()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
- when(zkmd.getDownloadUrl()).thenReturn("file://somewhere");
- String tierName = "coolTier";
- when(zkmd.getTier()).thenReturn(tierName);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
+ when(zkMetadata.getDownloadUrl()).thenReturn("file://somewhere");
+ when(zkMetadata.getTier()).thenReturn(TIER_NAME);
// No dataDir for coolTier, thus stay on default tier.
- BaseTableDataManager tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName,
- TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null), zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), localSegDir);
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, createTierIndexLoadingConfig(DEFAULT_TABLE_CONFIG));
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
// Configured dataDir for coolTier, thus move to new dir.
- tableConfig = createTableConfigWithTier(tierName, new File(TEMP_DIR, tierName));
- IndexLoadingConfig loadingCfg = TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, null);
- File segDirOnTier = tmgr.getSegmentDataDir(segName, tierName, loadingCfg.getTableConfig());
- assertFalse(segDirOnTier.exists());
- // Move segDir to new tier to see if addOrReplaceSegment() can load segDir from new tier directly.
- FileUtils.moveDirectory(localSegDir, segDirOnTier);
- tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName, loadingCfg, zkmd, null);
- llmd = new SegmentMetadataImpl(segDirOnTier);
- assertEquals(llmd.getTotalDocs(), 5);
- assertEquals(llmd.getIndexDir(), segDirOnTier);
+ File tierDataDir = tableDataManager.getSegmentDataDir(SEGMENT_NAME, TIER_NAME, TIER_TABLE_CONFIG);
+ assertFalse(tierDataDir.exists());
+ // Move segDir to new tier to see if addNewOnlineSegment() can load segDir from new tier directly.
+ FileUtils.moveDirectory(indexDir, tierDataDir);
+ tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, createTierIndexLoadingConfig(TIER_TABLE_CONFIG));
+ assertEquals(new SegmentMetadataImpl(tierDataDir).getTotalDocs(), 5);
}
@Test
- public void testAddOrReplaceSegmentUseBackupCopy()
+ public void testAddSegmentUseBackupCopy()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
- BaseTableDataManager tmgr = createTableManager();
- File backup = tmgr.getSegmentDataDir(segName + CommonConstants.Segment.SEGMENT_BACKUP_DIR_SUFFIX);
- localSegDir.renameTo(backup);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File backupDir =
+ tableDataManager.getSegmentDataDir(SEGMENT_NAME + CommonConstants.Segment.SEGMENT_BACKUP_DIR_SUFFIX);
+ assertTrue(indexDir.renameTo(backupDir));
- assertFalse(tmgr.getSegmentDataDir(segName).exists());
- tmgr.addOrReplaceSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
+ assertFalse(indexDir.exists());
+ tableDataManager.addNewOnlineSegment(zkMetadata, new IndexLoadingConfig());
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
}
@Test
- public void testAddOrReplaceSegmentStaleBackupCopy()
+ public void testAddSegmentStaleBackupCopy()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- SegmentZKMetadata zkmd = createRawSegment(tableConfig, segName, SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
- BaseTableDataManager tmgr = createTableManager();
// Create a local segment with fewer rows, making its CRC different from the raw segment.
// So that the raw segment is downloaded and loaded in the end.
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 3);
- File backup = tmgr.getSegmentDataDir(segName + CommonConstants.Segment.SEGMENT_BACKUP_DIR_SUFFIX);
- localSegDir.renameTo(backup);
+ BaseTableDataManager tableDataManager = createTableManager();
+ File indexDir = createSegment(SegmentVersion.v3, 3);
+ File backupDir =
+ tableDataManager.getSegmentDataDir(SEGMENT_NAME + CommonConstants.Segment.SEGMENT_BACKUP_DIR_SUFFIX);
+ assertTrue(indexDir.renameTo(backupDir));
- assertFalse(tmgr.getSegmentDataDir(segName).exists());
- tmgr.addOrReplaceSegment(segName, TableDataManagerTestUtils.createIndexLoadingConfig(), zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
+ assertFalse(indexDir.exists());
+ tableDataManager.addNewOnlineSegment(zkMetadata, new IndexLoadingConfig());
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 5);
}
@Test
- public void testAddOrReplaceSegmentUpConvertVersion()
+ public void testAddSegmentUpConvertVersion()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v1, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v1);
+ File indexDir = createSegment(SegmentVersion.v1, 5);
+ long crc = getCRC(indexDir);
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
// Require to use v3 format.
- IndexLoadingConfig idxCfg = TableDataManagerTestUtils.createIndexLoadingConfig();
- idxCfg.setSegmentVersion(SegmentVersion.v3);
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
+ indexLoadingConfig.setSegmentVersion(SegmentVersion.v3);
- BaseTableDataManager tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName, idxCfg, zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getVersion(), SegmentVersion.v3);
- assertEquals(llmd.getTotalDocs(), 5);
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ assertEquals(segmentMetadata.getVersion(), SegmentVersion.v3);
}
@Test
- public void testAddOrReplaceSegmentDownConvertVersion()
+ public void testAddSegmentDownConvertVersion()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
// Require to use v1 format.
- IndexLoadingConfig idxCfg = TableDataManagerTestUtils.createIndexLoadingConfig();
- idxCfg.setSegmentVersion(SegmentVersion.v1);
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
+ indexLoadingConfig.setSegmentVersion(SegmentVersion.v1);
- BaseTableDataManager tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName, idxCfg, zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
// The existing segment preprocessing logic doesn't down convert segment format.
- assertEquals(llmd.getVersion(), SegmentVersion.v3);
- assertEquals(llmd.getTotalDocs(), 5);
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ assertEquals(segmentMetadata.getVersion(), SegmentVersion.v3);
}
@Test
- public void testAddOrReplaceSegmentAddIndex()
+ public void testAddSegmentAddIndex()
throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
- String segName = "seg01";
- File localSegDir = createSegment(tableConfig, segName, SegmentVersion.v3, 5);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
- assertFalse(hasInvertedIndex(localSegDir, STRING_COLUMN, SegmentVersion.v3));
- assertFalse(hasInvertedIndex(localSegDir, LONG_COLUMN, SegmentVersion.v3));
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
+ assertFalse(hasInvertedIndex(indexDir, STRING_COLUMN));
+ assertFalse(hasInvertedIndex(indexDir, LONG_COLUMN));
// Make local and remote CRC same to skip downloading raw segment.
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getCrc()).thenReturn(segCrc);
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getCrc()).thenReturn(crc);
// Require to add indices.
- IndexLoadingConfig idxCfg = TableDataManagerTestUtils.createIndexLoadingConfig();
- idxCfg.setSegmentVersion(SegmentVersion.v3);
- idxCfg.setInvertedIndexColumns(new HashSet<>(Arrays.asList(STRING_COLUMN, LONG_COLUMN)));
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
+ indexLoadingConfig.setSegmentVersion(SegmentVersion.v3);
+ indexLoadingConfig.setInvertedIndexColumns(new HashSet<>(Arrays.asList(STRING_COLUMN, LONG_COLUMN)));
- BaseTableDataManager tmgr = createTableManager();
- tmgr.addOrReplaceSegment(segName, idxCfg, zkmd, null);
- assertTrue(tmgr.getSegmentDataDir(segName).exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(tmgr.getSegmentDataDir(segName));
- assertEquals(llmd.getTotalDocs(), 5);
- assertTrue(hasInvertedIndex(tmgr.getSegmentDataDir(segName), STRING_COLUMN, SegmentVersion.v3));
- assertTrue(hasInvertedIndex(tmgr.getSegmentDataDir(segName), LONG_COLUMN, SegmentVersion.v3));
+ BaseTableDataManager tableDataManager = createTableManager();
+ tableDataManager.addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+ assertEquals(tableDataManager.getSegmentDataDir(SEGMENT_NAME), indexDir);
+ assertTrue(indexDir.exists());
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
+ assertEquals(segmentMetadata.getTotalDocs(), 5);
+ assertTrue(hasInvertedIndex(indexDir, STRING_COLUMN));
+ assertTrue(hasInvertedIndex(indexDir, LONG_COLUMN));
}
@Test
public void testDownloadAndDecrypt()
throws Exception {
- File tempInput = new File(TEMP_DIR, "tmp.txt");
- FileUtils.write(tempInput, "this is from somewhere remote");
-
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getDownloadUrl()).thenReturn("file://" + tempInput.getAbsolutePath());
-
- BaseTableDataManager tmgr = createTableManager();
- File tempRootDir = tmgr.getTmpSegmentDataDir("test-download-decrypt");
-
- File tarFile = tmgr.downloadAndDecrypt("seg01", zkmd, tempRootDir);
- assertEquals(FileUtils.readFileToString(tarFile), "this is from somewhere remote");
-
- when(zkmd.getCrypterName()).thenReturn("fakePinotCrypter");
- tarFile = tmgr.downloadAndDecrypt("seg01", zkmd, tempRootDir);
- assertEquals(FileUtils.readFileToString(tarFile), "this is from somewhere remote");
+ File tempDir = new File(TEMP_DIR, "testDownloadAndDecrypt");
+ String fileName = "tmp.txt";
+ FileUtils.write(new File(tempDir, fileName), "this is from somewhere remote");
+ String tarFileName = SEGMENT_NAME + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION;
+ File tempTarFile = new File(TEMP_DIR, tarFileName);
+ TarGzCompressionUtils.createTarGzFile(tempDir, tempTarFile);
+
+ SegmentZKMetadata zkMetadata = mock(SegmentZKMetadata.class);
+ when(zkMetadata.getSegmentName()).thenReturn(SEGMENT_NAME);
+ when(zkMetadata.getDownloadUrl()).thenReturn("file://" + tempTarFile.getAbsolutePath());
+
+ BaseTableDataManager tableDataManager = createTableManager();
+ File indexDir = tableDataManager.downloadSegment(zkMetadata);
+ assertEquals(FileUtils.readFileToString(new File(indexDir, fileName)), "this is from somewhere remote");
+
+ FileUtils.deleteDirectory(indexDir);
+ when(zkMetadata.getCrypterName()).thenReturn("fakePinotCrypter");
+ indexDir = tableDataManager.downloadSegment(zkMetadata);
+ assertEquals(FileUtils.readFileToString(new File(indexDir, fileName)), "this is from somewhere remote");
FakePinotCrypter fakeCrypter = (FakePinotCrypter) PinotCrypterFactory.create("fakePinotCrypter");
- String parentDir = TABLE_NAME_WITH_TYPE + "/tmp/test-download-decrypt/";
- assertTrue(fakeCrypter._origFile.getAbsolutePath().endsWith(parentDir + "seg01.tar.gz.enc"));
- assertTrue(fakeCrypter._decFile.getAbsolutePath().endsWith(parentDir + "seg01.tar.gz"));
+ assertTrue(fakeCrypter._origFile.getAbsolutePath().endsWith(tarFileName + ".enc"));
+ assertTrue(fakeCrypter._decFile.getAbsolutePath().endsWith(tarFileName));
try {
// Set maxRetry to 0 to cause retry failure immediately.
Map<String, Object> properties = new HashMap<>();
properties.put(BaseSegmentFetcher.RETRY_COUNT_CONFIG_KEY, 0);
SegmentFetcherFactory.init(new PinotConfiguration(properties));
- tmgr.downloadAndDecrypt("seg01", zkmd, tempRootDir);
+ tableDataManager.downloadSegment(zkMetadata);
fail();
} catch (AttemptsExceededException e) {
assertEquals(e.getMessage(), "Operation failed after 0 attempts");
}
}
- // case 2: if the attempt to download from deep storage exceeds, invoke downloadFromPeers.
- @Test
- public void testDownloadAndDecryptPeerDownload()
- throws Exception {
- String backupCopyURI = mockRemoteCopy().toString();
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getDownloadUrl()).thenReturn(backupCopyURI);
-
- InstanceDataManagerConfig config = createDefaultInstanceDataManagerConfig();
- when(config.getSegmentPeerDownloadScheme()).thenReturn("http");
- BaseTableDataManager tmgr = spy(createTableManager(config));
- File tempRootDir = tmgr.getTmpSegmentDataDir("test-download-decrypt-peer");
-
- // As the case 2 description says, we need to mock the static method fetchAndDecryptSegmentToLocal to
- // throw the AttemptExceed exception; Due to the constraint that mockito static cannot do argument matching,
- // e.g., any(), we have to pass exact argument value when mocking fetchAndDecryptSegmentToLocal.
- // However, the second argument of File is internally created, which cannot be mocked.
- // Luckily, the File class's equal method only compares the path. Thus, we can create a file with identical path
- // and use it to mock the fetchAndDecryptSegmentToLocal
- File destFile = new File(tempRootDir, "seg01" + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
- doNothing().when(tmgr).downloadFromPeersWithoutStreaming("seg01", zkmd, destFile);
- try (MockedStatic<SegmentFetcherFactory> mockSegFactory = mockStatic(SegmentFetcherFactory.class)) {
- mockSegFactory.when(() -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(backupCopyURI, destFile, null))
- .thenThrow(new AttemptsExceededException("fake attempt exceeds exception"));
- tmgr.downloadAndDecrypt("seg01", zkmd, tempRootDir);
- }
- verify(tmgr, times(1)).downloadFromPeersWithoutStreaming("seg01", zkmd, destFile);
- }
-
@Test
public void testUntarAndMoveSegment()
throws IOException {
- BaseTableDataManager tmgr = createTableManager();
- File tempRootDir = tmgr.getTmpSegmentDataDir("test-untar-move");
+ BaseTableDataManager tableDataManager = createTableManager();
+ File tempRootDir = tableDataManager.getTmpSegmentDataDir("test-untar-move");
// All input and intermediate files are put in the tempRootDir.
- File tempTar = new File(tempRootDir, "seg01" + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
- File tempInputDir = new File(tempRootDir, "seg01_input");
+ File tempTar = new File(tempRootDir, SEGMENT_NAME + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ File tempInputDir = new File(tempRootDir, "input");
FileUtils.write(new File(tempInputDir, "tmp.txt"), "this is in segment dir");
TarGzCompressionUtils.createTarGzFile(tempInputDir, tempTar);
FileUtils.deleteQuietly(tempInputDir);
// The destination is the segment directory at the same level of tempRootDir.
- File indexDir = tmgr.untarAndMoveSegment("seg01", tempTar, tempRootDir);
- assertEquals(indexDir, tmgr.getSegmentDataDir("seg01"));
- assertEquals(FileUtils.readFileToString(new File(indexDir, "tmp.txt")), "this is in segment dir");
+ File untarredFile = tableDataManager.untarAndMoveSegment(SEGMENT_NAME, tempTar, tempRootDir);
+ assertEquals(untarredFile, tableDataManager.getSegmentDataDir(SEGMENT_NAME));
+ assertEquals(FileUtils.readFileToString(new File(untarredFile, "tmp.txt")), "this is in segment dir");
try {
- tmgr.untarAndMoveSegment("seg01", new File(tempRootDir, "unknown.txt"), TEMP_DIR);
+ tableDataManager.untarAndMoveSegment(SEGMENT_NAME, new File(tempRootDir, "unknown.txt"), TEMP_DIR);
fail();
} catch (Exception e) {
// expected.
@@ -689,6 +643,7 @@ public class BaseTableDataManagerTest {
public void decrypt(File origFile, File decFile) {
_origFile = origFile;
_decFile = decFile;
+ origFile.renameTo(decFile);
}
}
@@ -697,15 +652,10 @@ public class BaseTableDataManagerTest {
}
private static OfflineTableDataManager createTableManager(InstanceDataManagerConfig instanceDataManagerConfig) {
- return createTableManager(instanceDataManagerConfig, mock(HelixManager.class));
- }
-
- private static OfflineTableDataManager createTableManager(InstanceDataManagerConfig instanceDataManagerConfig,
- HelixManager helixManager) {
- TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ HelixManager helixManager = mock(HelixManager.class);
+ SegmentLocks segmentLocks = new SegmentLocks();
OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
- tableDataManager.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
- tableDataManager.start();
+ tableDataManager.init(instanceDataManagerConfig, helixManager, segmentLocks, DEFAULT_TABLE_CONFIG, null, null);
return tableDataManager;
}
@@ -715,24 +665,14 @@ public class BaseTableDataManagerTest {
return config;
}
- private static SegmentZKMetadata createRawSegment(TableConfig tableConfig, String segName, SegmentVersion segVer,
- int rowCnt)
- throws Exception {
- File localSegDir = createSegment(tableConfig, segName, segVer, rowCnt);
- return TableDataManagerTestUtils.makeRawSegment(segName, localSegDir,
- new File(TEMP_DIR, segName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION), true);
- }
-
- private static File createSegment(TableConfig tableConfig, String segName, SegmentVersion segVer, int rowCnt)
+ private static File createSegment(SegmentVersion segmentVersion, int numRows)
throws Exception {
- Schema schema = new Schema.SchemaBuilder().addSingleValueDimension(STRING_COLUMN, FieldSpec.DataType.STRING)
- .addMetric(LONG_COLUMN, FieldSpec.DataType.LONG).build();
- SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, schema);
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(DEFAULT_TABLE_CONFIG, SCHEMA);
config.setOutDir(TABLE_DATA_DIR.getAbsolutePath());
- config.setSegmentName(segName);
- config.setSegmentVersion(segVer);
+ config.setSegmentName(SEGMENT_NAME);
+ config.setSegmentVersion(segmentVersion);
List<GenericRow> rows = new ArrayList<>(3);
- for (int i = 0; i < rowCnt; i++) {
+ for (int i = 0; i < numRows; i++) {
GenericRow row = new GenericRow();
row.putValue(STRING_COLUMN, STRING_VALUES[i]);
row.putValue(LONG_COLUMN, LONG_VALUES[i]);
@@ -741,33 +681,64 @@ public class BaseTableDataManagerTest {
SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
driver.init(config, new GenericRowRecordReader(rows));
driver.build();
- return new File(TABLE_DATA_DIR, segName);
+ return new File(TABLE_DATA_DIR, SEGMENT_NAME);
}
- private static boolean hasInvertedIndex(File segDir, String colName, SegmentVersion segVer)
+ private static SegmentZKMetadata createRawSegment(SegmentVersion segmentVersion, int numRows)
+ throws Exception {
+ File indexDir = createSegment(segmentVersion, numRows);
+ return makeRawSegment(indexDir, new File(TEMP_DIR, SEGMENT_NAME + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION),
+ true);
+ }
+
+ private static SegmentZKMetadata makeRawSegment(File indexDir, File rawSegmentFile, boolean deleteIndexDir)
+ throws Exception {
+ long crc = getCRC(indexDir);
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ TarGzCompressionUtils.createTarGzFile(indexDir, rawSegmentFile);
+ zkMetadata.setDownloadUrl("file://" + rawSegmentFile.getAbsolutePath());
+ zkMetadata.setCrc(crc);
+ if (deleteIndexDir) {
+ FileUtils.deleteQuietly(indexDir);
+ }
+ return zkMetadata;
+ }
+
+ private static long getCRC(File indexDir)
throws IOException {
- File parentDir = segDir;
- if (segVer == SegmentVersion.v3) {
- parentDir = new File(segDir, "v3");
+ File creationMetaFile = SegmentDirectoryPaths.findCreationMetaFile(indexDir);
+ assertNotNull(creationMetaFile);
+ try (DataInputStream in = new DataInputStream(new FileInputStream(creationMetaFile))) {
+ return in.readLong();
}
- File idxMapFile = new File(parentDir, V1Constants.INDEX_MAP_FILE_NAME);
- return FileUtils.readFileToString(idxMapFile).contains(colName + ".inverted_index");
}
- private TableConfig createTableConfigWithTier(String tierName, File dataDir) {
- return new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setTierConfigList(
- Collections.singletonList(new TierConfig(tierName, TierFactory.TIME_SEGMENT_SELECTOR_TYPE, "3d", null,
- TierFactory.PINOT_SERVER_STORAGE_TYPE, "tag_OFFLINE", null,
- Collections.singletonMap("dataDir", dataDir.getAbsolutePath())))).build();
+ private IndexLoadingConfig createTierIndexLoadingConfig(TableConfig tableConfig) {
+ InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
+ when(instanceDataManagerConfig.getSegmentDirectoryLoader()).thenReturn(TIER_SEGMENT_DIRECTORY_LOADER);
+ when(instanceDataManagerConfig.getConfig()).thenReturn(new PinotConfiguration());
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(instanceDataManagerConfig, tableConfig, null);
+ indexLoadingConfig.setTableDataDir(TEMP_DIR.getAbsolutePath() + File.separator + tableConfig.getTableName());
+ indexLoadingConfig.setInstanceTierConfigs(Map.of());
+ indexLoadingConfig.setSegmentTier(TIER_NAME);
+ return indexLoadingConfig;
}
- private static URI mockRemoteCopy()
- throws IOException, URISyntaxException {
- File tempInput = new File(TEMP_DIR, "tmp.txt");
- FileUtils.write(tempInput, "this is from somewhere remote");
+ private ImmutableSegmentDataManager createImmutableSegmentDataManager(String segmentName, long crc) {
+ ImmutableSegmentDataManager segmentDataManager = mock(ImmutableSegmentDataManager.class);
+ when(segmentDataManager.getSegmentName()).thenReturn(segmentName);
+ ImmutableSegment immutableSegment = mock(ImmutableSegment.class);
+ when(segmentDataManager.getSegment()).thenReturn(immutableSegment);
+ SegmentMetadata segmentMetadata = mock(SegmentMetadata.class);
+ when(immutableSegment.getSegmentMetadata()).thenReturn(segmentMetadata);
+ when(segmentMetadata.getCrc()).thenReturn(Long.toString(crc));
+ return segmentDataManager;
+ }
- String backupCopyURI = "file://" + tempInput.getAbsolutePath();
- URI uri = new URI(backupCopyURI);
- return uri;
+ private static boolean hasInvertedIndex(File indexDir, String columnName)
+ throws IOException {
+ File indexMapFile =
+ new File(new File(indexDir, SegmentDirectoryPaths.V3_SUBDIRECTORY_NAME), V1Constants.INDEX_MAP_FILE_NAME);
+ return FileUtils.readFileToString(indexMapFile, StandardCharsets.UTF_8).contains(columnName + ".inverted_index");
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableDataManagerTestUtils.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableDataManagerTestUtils.java
deleted file mode 100644
index 510e6c1232..0000000000
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/TableDataManagerTestUtils.java
+++ /dev/null
@@ -1,107 +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.pinot.core.data.manager;
-
-import java.io.DataInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import javax.annotation.Nullable;
-import org.apache.commons.io.FileUtils;
-import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.utils.TarGzCompressionUtils;
-import org.apache.pinot.common.utils.fetcher.BaseSegmentFetcher;
-import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
-import org.apache.pinot.segment.spi.V1Constants;
-import org.apache.pinot.segment.spi.creator.SegmentVersion;
-import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
-import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.crypt.PinotCrypterFactory;
-import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.utils.ReadMode;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-public class TableDataManagerTestUtils {
- private TableDataManagerTestUtils() {
- }
-
- public static long getCRC(File segDir, SegmentVersion segVer)
- throws IOException {
- File parentDir = segDir;
- if (segVer == SegmentVersion.v3) {
- parentDir = new File(segDir, "v3");
- }
- File crcFile = new File(parentDir, V1Constants.SEGMENT_CREATION_META);
- try (DataInputStream ds = new DataInputStream(new FileInputStream(crcFile))) {
- return ds.readLong();
- }
- }
-
- public static SegmentZKMetadata makeRawSegment(String segName, File localSegDir, File rawSegDir,
- boolean deleteLocalSegDir)
- throws Exception {
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
- SegmentZKMetadata zkmd = new SegmentZKMetadata(segName);
- TarGzCompressionUtils.createTarGzFile(localSegDir, rawSegDir);
- zkmd.setDownloadUrl("file://" + rawSegDir.getAbsolutePath());
- zkmd.setCrc(segCrc);
- if (deleteLocalSegDir) {
- FileUtils.deleteQuietly(localSegDir);
- }
- return zkmd;
- }
-
- public static void initSegmentFetcher()
- throws Exception {
- Map<String, Object> properties = new HashMap<>();
- properties.put(BaseSegmentFetcher.RETRY_COUNT_CONFIG_KEY, 3);
- properties.put(BaseSegmentFetcher.RETRY_WAIT_MS_CONFIG_KEY, 100);
- properties.put(BaseSegmentFetcher.RETRY_DELAY_SCALE_FACTOR_CONFIG_KEY, 5);
- SegmentFetcherFactory.init(new PinotConfiguration(properties));
-
- // Setup crypter
- properties.put("class.fakePinotCrypter", BaseTableDataManagerTest.FakePinotCrypter.class.getName());
- PinotCrypterFactory.init(new PinotConfiguration(properties));
- }
-
- public static IndexLoadingConfig createIndexLoadingConfig(String segDirLoader, TableConfig tableConfig,
- @Nullable Schema schema) {
- InstanceDataManagerConfig idmc = mock(InstanceDataManagerConfig.class);
- when(idmc.getSegmentDirectoryLoader()).thenReturn(segDirLoader);
- when(idmc.getConfig()).thenReturn(new PinotConfiguration());
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(idmc, tableConfig, schema);
- indexLoadingConfig.setSegmentVersion(SegmentVersion.v3);
- indexLoadingConfig.setReadMode(ReadMode.mmap);
- return indexLoadingConfig;
- }
-
- public static IndexLoadingConfig createIndexLoadingConfig() {
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();
- indexLoadingConfig.setSegmentVersion(SegmentVersion.v3);
- indexLoadingConfig.setReadMode(ReadMode.mmap);
- return indexLoadingConfig;
- }
-}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManagerTest.java
index 8c865ba925..fd6769a0ad 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManagerTest.java
@@ -35,10 +35,12 @@ import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.utils.SchemaUtils;
import org.apache.pinot.common.utils.config.TableConfigUtils;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentCreationDriverFactory;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.index.loader.LoaderTest;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
@@ -96,8 +98,8 @@ public class DimensionTableDataManagerTest {
assertNotNull(schemaPathUrl);
assertNotNull(configPathUrl);
File csvFile = new File(dataPathUrl.getFile());
- Schema schema = createSchema(new File(schemaPathUrl.getFile()));
TableConfig tableConfig = createTableConfig(new File(configPathUrl.getFile()));
+ Schema schema = createSchema(new File(schemaPathUrl.getFile()));
// create segment
File tableDataDir = new File(TEMP_DIR, OFFLINE_TABLE_NAME);
@@ -111,7 +113,7 @@ public class DimensionTableDataManagerTest {
String segmentName = driver.getSegmentName();
_indexDir = new File(tableDataDir, segmentName);
- _indexLoadingConfig = new IndexLoadingConfig();
+ _indexLoadingConfig = new IndexLoadingConfig(tableConfig, schema);
_segmentMetadata = new SegmentMetadataImpl(_indexDir);
_segmentZKMetadata = new SegmentZKMetadata(segmentName);
_segmentZKMetadata.setCrc(Long.parseLong(_segmentMetadata.getCrc()));
@@ -147,7 +149,7 @@ public class DimensionTableDataManagerTest {
TableConfig tableConfig = getTableConfig(false, false);
DimensionTableDataManager tableDataManager =
DimensionTableDataManager.createInstanceByTableName(OFFLINE_TABLE_NAME);
- tableDataManager.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
+ tableDataManager.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), tableConfig, null, null);
tableDataManager.start();
return tableDataManager;
}
@@ -169,13 +171,13 @@ public class DimensionTableDataManagerTest {
assertEquals(tableDataManager, returnedManager, "Manager should return already created instance");
// assert that segments are released after loading data
- tableDataManager.addSegment(_indexDir, _indexLoadingConfig);
+ tableDataManager.addSegment(ImmutableSegmentLoader.load(_indexDir, _indexLoadingConfig));
for (SegmentDataManager segmentManager : returnedManager.acquireAllSegments()) {
assertEquals(segmentManager.getReferenceCount() - 1, // Subtract this acquisition
1, // Default ref count
"Reference counts should be same before and after segment loading.");
returnedManager.releaseSegment(segmentManager);
- returnedManager.removeSegment(segmentManager.getSegmentName());
+ returnedManager.offloadSegment(segmentManager.getSegmentName());
}
// try fetching non-existent table
@@ -197,7 +199,7 @@ public class DimensionTableDataManagerTest {
GenericRow resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
assertNull(resp, "Response should be null if no segment is loaded");
- tableDataManager.addSegment(_indexDir, _indexLoadingConfig);
+ tableDataManager.addSegment(ImmutableSegmentLoader.load(_indexDir, _indexLoadingConfig));
// Confirm table is loaded and available for lookup
resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
@@ -220,7 +222,7 @@ public class DimensionTableDataManagerTest {
assertEquals(segmentManagers.size(), 1, "Should have exactly one segment manager");
SegmentDataManager segMgr = segmentManagers.get(0);
String segmentName = segMgr.getSegmentName();
- tableDataManager.removeSegment(segmentName);
+ tableDataManager.offloadSegment(segmentName);
// confirm table is cleaned up
resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
assertNull(resp, "Response should be null if no segment is loaded");
@@ -235,8 +237,7 @@ public class DimensionTableDataManagerTest {
SchemaUtils.toZNRecord(getSchema()));
when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
DimensionTableDataManager tableDataManager = makeTableDataManager(helixManager);
-
- tableDataManager.addSegment(_indexDir, _indexLoadingConfig);
+ tableDataManager.addSegment(ImmutableSegmentLoader.load(_indexDir, _indexLoadingConfig));
// Confirm table is loaded and available for lookup
GenericRow resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
@@ -283,7 +284,7 @@ public class DimensionTableDataManagerTest {
GenericRow resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
assertNull(resp, "Response should be null if no segment is loaded");
- tableDataManager.addSegment(_indexDir, _indexLoadingConfig);
+ tableDataManager.addSegment(ImmutableSegmentLoader.load(_indexDir, _indexLoadingConfig));
// Confirm table is loaded and available for lookup
resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
@@ -306,7 +307,7 @@ public class DimensionTableDataManagerTest {
assertEquals(segmentManagers.size(), 1, "Should have exactly one segment manager");
SegmentDataManager segMgr = segmentManagers.get(0);
String segmentName = segMgr.getSegmentName();
- tableDataManager.removeSegment(segmentName);
+ tableDataManager.offloadSegment(segmentName);
// confirm table is cleaned up
resp = tableDataManager.lookupRowByPrimaryKey(new PrimaryKey(new String[]{"SF"}));
assertNull(resp, "Response should be null if no segment is loaded");
@@ -329,7 +330,7 @@ public class DimensionTableDataManagerTest {
assertNull(resp, "Response should be null if no segment is loaded");
try {
- tableDataManager.addSegment(_indexDir, _indexLoadingConfig);
+ tableDataManager.addSegment(ImmutableSegmentLoader.load(_indexDir, _indexLoadingConfig));
fail("Should error out when ErrorOnDuplicatePrimaryKey is configured to true");
} catch (Exception e) {
// expected;
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTrackerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTrackerTest.java
index 177fe33269..b6c61e5f50 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTrackerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTrackerTest.java
@@ -24,20 +24,22 @@ import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.spi.metrics.PinotMetricUtils;
import org.testng.Assert;
import org.testng.annotations.Test;
-public class IngestionDelayTrackerTest {
+import static org.mockito.Mockito.mock;
+
+public class IngestionDelayTrackerTest {
+ private static final String REALTIME_TABLE_NAME = "testTable_REALTIME";
private static final int TIMER_THREAD_TICK_INTERVAL_MS = 100;
+ private final ServerMetrics _serverMetrics = mock(ServerMetrics.class);
+ private final RealtimeTableDataManager _realtimeTableDataManager = mock(RealtimeTableDataManager.class);
+
private IngestionDelayTracker createTracker() {
- ServerMetrics serverMetrics = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
- RealtimeTableDataManager realtimeTableDataManager = new RealtimeTableDataManager(null);
IngestionDelayTracker ingestionDelayTracker =
- new IngestionDelayTracker(serverMetrics, "dummyTable_RT",
- realtimeTableDataManager, () -> true);
+ new IngestionDelayTracker(_serverMetrics, REALTIME_TABLE_NAME, _realtimeTableDataManager, () -> true);
// With no samples, the time reported must be zero
Assert.assertEquals(ingestionDelayTracker.getPartitionIngestionDelayMs(0), 0);
return ingestionDelayTracker;
@@ -45,23 +47,18 @@ public class IngestionDelayTrackerTest {
@Test
public void testTrackerConstructors() {
- ServerMetrics serverMetrics = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
- RealtimeTableDataManager realtimeTableDataManager = new RealtimeTableDataManager(null);
// Test regular constructor
IngestionDelayTracker ingestionDelayTracker =
- new IngestionDelayTracker(serverMetrics, "dummyTable_RT",
- realtimeTableDataManager, () -> true);
+ new IngestionDelayTracker(_serverMetrics, REALTIME_TABLE_NAME, _realtimeTableDataManager, () -> true);
Assert.assertEquals(ingestionDelayTracker.getPartitionIngestionDelayMs(0), 0);
ingestionDelayTracker.shutdown();
// Test constructor with timer arguments
- ingestionDelayTracker =
- new IngestionDelayTracker(serverMetrics, "dummyTable_RT",
- realtimeTableDataManager, TIMER_THREAD_TICK_INTERVAL_MS, () -> true);
+ ingestionDelayTracker = new IngestionDelayTracker(_serverMetrics, REALTIME_TABLE_NAME, _realtimeTableDataManager,
+ TIMER_THREAD_TICK_INTERVAL_MS, () -> true);
Assert.assertEquals(ingestionDelayTracker.getPartitionIngestionDelayMs(0), 0);
// Test bad timer args to the constructor
try {
- new IngestionDelayTracker(serverMetrics, "dummyTable_RT",
- realtimeTableDataManager, 0, () -> true);
+ new IngestionDelayTracker(_serverMetrics, REALTIME_TABLE_NAME, _realtimeTableDataManager, 0, () -> true);
Assert.fail("Must have asserted due to invalid arguments"); // Constructor must assert
} catch (Exception e) {
if ((e instanceof NullPointerException) || !(e instanceof RuntimeException)) {
@@ -140,8 +137,8 @@ public class IngestionDelayTrackerTest {
ZoneId zoneId = ZoneId.systemDefault();
Clock clock = Clock.fixed(now, zoneId);
ingestionDelayTracker.setClock(clock);
- ingestionDelayTracker.updateIngestionDelay((clock.millis() - partition0Delay0),
- (clock.millis() - partition0Delay0), partition0);
+ ingestionDelayTracker.updateIngestionDelay((clock.millis() - partition0Delay0), (clock.millis() - partition0Delay0),
+ partition0);
Assert.assertEquals(ingestionDelayTracker.getPartitionIngestionDelayMs(partition0), partition0Delay0);
Assert.assertEquals(ingestionDelayTracker.getPartitionEndToEndIngestionDelayMs(partition0), partition0Delay0);
// Advance clock and test aging
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index 088e270a32..910edf7d9b 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.data.manager.realtime;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
@@ -31,6 +32,7 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
import java.util.function.Supplier;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
@@ -49,6 +51,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory;
import org.apache.pinot.segment.local.segment.creator.Fixtures;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
@@ -102,6 +105,7 @@ public class RealtimeSegmentDataManagerTest {
tableConfig.getTableName());
RealtimeTableDataManager tableDataManager = mock(RealtimeTableDataManager.class);
when(tableDataManager.getInstanceId()).thenReturn(instanceId);
+ when(tableDataManager.getSegmentLock(any())).thenReturn(mock(Lock.class));
RealtimeSegmentStatsHistory statsHistory = mock(RealtimeSegmentStatsHistory.class);
when(statsHistory.getEstimatedCardinality(anyString())).thenReturn(200);
when(statsHistory.getEstimatedAvgColSize(anyString())).thenReturn(32);
@@ -193,7 +197,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._commitSegmentCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.HOLDING);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
// Test that we go to commit when the controller responds commit after 2 holds.
@@ -224,7 +228,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.COMMITTED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -244,7 +248,7 @@ public class RealtimeSegmentDataManagerTest {
consumer.run();
Assert.assertTrue(segmentDataManager._buildSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.ERROR);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
// Test hold, catchup. hold, commit
@@ -287,7 +291,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.COMMITTED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -334,7 +338,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.COMMITTED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -381,7 +385,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.COMMITTED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -405,7 +409,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.DISCARDED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -430,7 +434,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
Assert.assertFalse(segmentDataManager._commitSegmentCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.RETAINED);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -458,7 +462,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager._commitSegmentCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), RealtimeSegmentDataManager.State.HOLDING);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -471,7 +475,7 @@ public class RealtimeSegmentDataManagerTest {
segmentDataManager._postConsumeStoppedCalled = false;
consumer.run();
Assert.assertTrue(segmentDataManager._postConsumeStoppedCalled);
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
// Tests to go online from consuming state
@@ -486,92 +490,76 @@ public class RealtimeSegmentDataManagerTest {
final LongMsgOffset finalOffset = new LongMsgOffset(finalOffsetValue);
metadata.setEndOffset(finalOffset.toString());
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.COMMITTED);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.RETAINED);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.DISCARDED);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.ERROR);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
// If holding, but we have overshot the expected final offset, the download and replace
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.HOLDING);
segmentDataManager.setCurrentOffset(finalOffsetValue + 1);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
// If catching up, but we have overshot the expected final offset, the download and replace
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
segmentDataManager.setCurrentOffset(finalOffsetValue + 1);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
// If catching up, but we did not get to the final offset, then download and replace
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
segmentDataManager._consumeOffsets.add(new LongMsgOffset(finalOffsetValue - 1));
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
// But then if we get to the exact offset, we get to build and replace, not download
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
segmentDataManager._consumeOffsets.add(finalOffset);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
- segmentDataManager.destroy();
}
}
@@ -579,8 +567,7 @@ public class RealtimeSegmentDataManagerTest {
public void testEndCriteriaChecking()
throws Exception {
// test reaching max row limit
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
segmentDataManager.setNumRowsIndexed(Fixtures.MAX_ROWS_IN_SEGMENT - 1);
@@ -588,11 +575,9 @@ public class RealtimeSegmentDataManagerTest {
segmentDataManager.setNumRowsIndexed(Fixtures.MAX_ROWS_IN_SEGMENT);
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
Assert.assertEquals(segmentDataManager.getStopReason(), SegmentCompletionProtocol.REASON_ROW_LIMIT);
- segmentDataManager.destroy();
}
// test reaching max time limit
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
// We should still get false because there is no messages fetched
@@ -603,11 +588,9 @@ public class RealtimeSegmentDataManagerTest {
segmentDataManager._timeSupplier.add(TimeUnit.HOURS.toMillis(1));
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
Assert.assertEquals(segmentDataManager.getStopReason(), SegmentCompletionProtocol.REASON_TIME_LIMIT);
- segmentDataManager.destroy();
}
// In catching up state, test reaching final offset
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
final long finalOffset = START_OFFSET_VALUE + 100;
segmentDataManager.setFinalOffset(finalOffset);
@@ -615,11 +598,9 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
segmentDataManager.setCurrentOffset(finalOffset);
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
- segmentDataManager.destroy();
}
// In catching up state, test reaching final offset ignoring time
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._timeSupplier.add(Fixtures.MAX_TIME_FOR_SEGMENT_CLOSE_MS);
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
final long finalOffset = START_OFFSET_VALUE + 100;
@@ -628,12 +609,10 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
segmentDataManager.setCurrentOffset(finalOffset);
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
- segmentDataManager.destroy();
}
// When we go from consuming to online state, time and final offset matter.
// Case 1. We have reached final offset.
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._timeSupplier.add(1);
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CONSUMING_TO_ONLINE);
segmentDataManager.setConsumeEndTime(segmentDataManager._timeSupplier.get() + 10);
@@ -643,11 +622,9 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
segmentDataManager.setCurrentOffset(finalOffset);
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
- segmentDataManager.destroy();
}
// Case 2. We have reached time limit.
- {
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager();
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CONSUMING_TO_ONLINE);
final long endTime = segmentDataManager._timeSupplier.get() + 10;
segmentDataManager.setConsumeEndTime(endTime);
@@ -658,7 +635,6 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached());
segmentDataManager._timeSupplier.set(endTime);
Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached());
- segmentDataManager.destroy();
}
}
@@ -702,7 +678,7 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertTrue(segmentTarFile.exists());
Assert.assertTrue(segmentDataManager.invokeCommit());
Assert.assertFalse(segmentTarFile.exists());
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
// If commit fails, and we still have the file, make sure that we remove the file when we go
@@ -736,7 +712,7 @@ public class RealtimeSegmentDataManagerTest {
segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.HOLDING);
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertFalse(segmentTarFile.exists());
- segmentDataManager.destroy();
+ segmentDataManager.close();
}
@Test
@@ -765,7 +741,7 @@ public class RealtimeSegmentDataManagerTest {
TestUtils.waitForCondition(aVoid -> {
if (firstSemaphore.hasQueuedThreads()) {
// Once verified the second segment gets blocked, release the semaphore.
- firstSegmentDataManager.destroy();
+ firstSegmentDataManager.close();
return true;
} else {
return false;
@@ -782,12 +758,12 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertEquals(secondSemaphore.availablePermits(), 0);
Assert.assertFalse(secondSemaphore.hasQueuedThreads());
- // Call destroy method the 2nd time on the first segment manager, the permits in semaphore won't increase.
- firstSegmentDataManager.destroy();
+ // Call offload method the 2nd time on the first segment manager, the permits in semaphore won't increase.
+ firstSegmentDataManager.close();
Assert.assertEquals(firstSegmentDataManager.getPartitionGroupConsumerSemaphore().availablePermits(), 0);
// The permit finally gets released in the Semaphore.
- secondSegmentDataManager.get().destroy();
+ secondSegmentDataManager.get().close();
Assert.assertEquals(secondSegmentDataManager.get().getPartitionGroupConsumerSemaphore().availablePermits(), 1);
}
@@ -804,7 +780,8 @@ public class RealtimeSegmentDataManagerTest {
InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManager tableDataManager =
- new TableDataManagerProvider(instanceDataManagerConfig).getTableDataManager(tableConfig, helixManager);
+ new TableDataManagerProvider(instanceDataManagerConfig, helixManager, new SegmentLocks()).getTableDataManager(
+ tableConfig);
tableDataManager.start();
tableDataManager.shutDown();
Assert.assertFalse(SegmentBuildTimeLeaseExtender.isExecutorShutdown());
@@ -826,24 +803,23 @@ public class RealtimeSegmentDataManagerTest {
return now + TimeUnit.MINUTES.toMillis(segmentTimeThresholdMins + 1);
}
};
- FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(true, timeSupplier,
- String.valueOf(FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS * 2), segmentTimeThresholdMins + "m", null);
- segmentDataManager._stubConsumeLoop = false;
- segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(true, timeSupplier,
+ String.valueOf(FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS * 2), segmentTimeThresholdMins + "m", null)) {
+ segmentDataManager._stubConsumeLoop = false;
+ segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
- RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
- final LongMsgOffset endOffset =
- new LongMsgOffset(START_OFFSET_VALUE + FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
- segmentDataManager._consumeOffsets.add(endOffset);
- final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
- new SegmentCompletionProtocol.Response.Params().withStatus(
- SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)
- .withStreamPartitionMsgOffset(endOffset.toString()));
- segmentDataManager._responses.add(response);
+ RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
+ final LongMsgOffset endOffset =
+ new LongMsgOffset(START_OFFSET_VALUE + FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
+ segmentDataManager._consumeOffsets.add(endOffset);
+ final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params().withStatus(
+ SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)
+ .withStreamPartitionMsgOffset(endOffset.toString()));
+ segmentDataManager._responses.add(response);
- consumer.run();
+ consumer.run();
- try {
// millis() is called first in run before consumption, then once for each batch and once for each message in
// the batch, then once more when metrics are updated after each batch is processed and then 4 more times in
// run() after consume loop
@@ -854,8 +830,6 @@ public class RealtimeSegmentDataManagerTest {
FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
Assert.assertEquals(segmentDataManager.getSegment().getSegmentMetadata().getTotalDocs(),
FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
- } finally {
- segmentDataManager.destroy();
}
}
@@ -864,25 +838,23 @@ public class RealtimeSegmentDataManagerTest {
throws Exception {
final int segmentTimeThresholdMins = 10;
TimeSupplier timeSupplier = new TimeSupplier();
- FakeRealtimeSegmentDataManager segmentDataManager =
- createFakeSegmentManager(true, timeSupplier, String.valueOf(FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS),
- segmentTimeThresholdMins + "m", null);
- segmentDataManager._stubConsumeLoop = false;
- segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+ try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(true, timeSupplier,
+ String.valueOf(FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS), segmentTimeThresholdMins + "m", null)) {
+ segmentDataManager._stubConsumeLoop = false;
+ segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
- RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
- final LongMsgOffset endOffset =
- new LongMsgOffset(START_OFFSET_VALUE + FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
- segmentDataManager._consumeOffsets.add(endOffset);
- final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
- new SegmentCompletionProtocol.Response.Params().withStatus(
- SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)
- .withStreamPartitionMsgOffset(endOffset.toString()));
- segmentDataManager._responses.add(response);
+ RealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer();
+ final LongMsgOffset endOffset =
+ new LongMsgOffset(START_OFFSET_VALUE + FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
+ segmentDataManager._consumeOffsets.add(endOffset);
+ final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params().withStatus(
+ SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)
+ .withStreamPartitionMsgOffset(endOffset.toString()));
+ segmentDataManager._responses.add(response);
- consumer.run();
+ consumer.run();
- try {
// millis() is called first in run before consumption, then once for each batch and once for each message in
// the batch, then once for metrics updates and then 4 more times in run() after consume loop
Assert.assertEquals(timeSupplier._timeCheckCounter.get(), FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS + 6);
@@ -892,8 +864,6 @@ public class RealtimeSegmentDataManagerTest {
FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
Assert.assertEquals(segmentDataManager.getSegment().getSegmentMetadata().getTotalDocs(),
FakeStreamConfigUtils.SEGMENT_FLUSH_THRESHOLD_ROWS);
- } finally {
- segmentDataManager.destroy();
}
}
@@ -916,7 +886,8 @@ public class RealtimeSegmentDataManagerTest {
}
}
- public static class FakeRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
+ // Implementing Closeable to make it easier to use in try-with-resources
+ public static class FakeRealtimeSegmentDataManager extends RealtimeSegmentDataManager implements Closeable {
public Field _state;
public Field _shouldStop;
@@ -985,7 +956,8 @@ public class RealtimeSegmentDataManagerTest {
return getSegmentBuildDescriptor();
}
- public boolean invokeCommit() {
+ public boolean invokeCommit()
+ throws Exception {
return super.commitSegment("dummyUrl");
}
@@ -1175,5 +1147,11 @@ public class RealtimeSegmentDataManagerTest {
Assert.fail();
}
}
+
+ @Override
+ public void close() {
+ offload();
+ destroy();
+ }
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
index bcf75a2cd0..ea266df007 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
@@ -18,254 +18,24 @@
*/
package org.apache.pinot.core.data.manager.realtime;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.commons.io.FileUtils;
-import org.apache.helix.AccessOption;
-import org.apache.helix.HelixManager;
-import org.apache.helix.store.zk.ZkHelixPropertyStore;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.SchemaUtils;
-import org.apache.pinot.common.utils.TarGzCompressionUtils;
-import org.apache.pinot.common.utils.config.TableConfigUtils;
-import org.apache.pinot.core.data.manager.TableDataManagerTestUtils;
-import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
-import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
-import org.apache.pinot.segment.spi.creator.SegmentVersion;
-import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
-import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.data.readers.GenericRow;
-import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
-import org.apache.pinot.util.TestUtils;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormat;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import static org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
public class RealtimeTableDataManagerTest {
- private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "RealtimeTableDataManagerTest");
- private static final String TABLE_NAME = "table01";
- private static final String TABLE_NAME_WITH_TYPE = "table01_REALTIME";
- private static final File TABLE_DATA_DIR = new File(TEMP_DIR, TABLE_NAME_WITH_TYPE);
- private static final String STRING_COLUMN = "col1";
- private static final String[] STRING_VALUES = {"A", "D", "E", "B", "C"};
- private static final String LONG_COLUMN = "col2";
- private static final long[] LONG_VALUES = {10000L, 20000L, 50000L, 40000L, 30000L};
-
- @BeforeClass
- public void setUp() {
- ServerMetrics.register(mock(ServerMetrics.class));
- }
-
- @BeforeMethod
- public void setUpMethod()
- throws Exception {
- TestUtils.ensureDirectoriesExistAndEmpty(TEMP_DIR);
- TableDataManagerTestUtils.initSegmentFetcher();
- }
-
- @AfterMethod
- public void tearDownMethod()
- throws Exception {
- FileUtils.deleteDirectory(TEMP_DIR);
- }
-
- @Test
- public void testAddSegmentUseBackupCopy()
- throws Exception {
- InstanceDataManagerConfig instanceDataManagerConfig = createInstanceDataManagerConfig();
- ZkHelixPropertyStore<ZNRecord> propertyStore = mock(ZkHelixPropertyStore.class);
- TableConfig tableConfig = setupTableConfig(propertyStore);
- Schema schema = setupSchema(propertyStore);
- HelixManager helixManager = mock(HelixManager.class);
- when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
- RealtimeTableDataManager tmgr = new RealtimeTableDataManager(null);
- tmgr.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
-
- // Create a dummy local segment.
- String segName = "seg01";
- SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segName);
- segmentZKMetadata.setStatus(Status.DONE);
- File localSegDir = createSegment(tableConfig, schema, segName);
- long segCrc = TableDataManagerTestUtils.getCRC(localSegDir, SegmentVersion.v3);
- segmentZKMetadata.setCrc(segCrc);
-
- // Move the segment to the backup location.
- File backup = new File(TABLE_DATA_DIR, segName + CommonConstants.Segment.SEGMENT_BACKUP_DIR_SUFFIX);
- localSegDir.renameTo(backup);
- assertEquals(localSegDir, new File(TABLE_DATA_DIR, segName));
- assertFalse(localSegDir.exists());
- IndexLoadingConfig indexLoadingConfig =
- TableDataManagerTestUtils.createIndexLoadingConfig("default", tableConfig, schema);
- tmgr.addSegment(segName, indexLoadingConfig, segmentZKMetadata);
- // Segment data is put back the default location, and backup location is deleted.
- assertTrue(localSegDir.exists());
- assertFalse(backup.exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(new File(TABLE_DATA_DIR, segName));
- assertEquals(llmd.getTotalDocs(), 5);
- }
-
- @Test
- public void testAddSegmentNoBackupCopy()
- throws Exception {
- InstanceDataManagerConfig instanceDataManagerConfig = createInstanceDataManagerConfig();
- ZkHelixPropertyStore<ZNRecord> propertyStore = mock(ZkHelixPropertyStore.class);
- TableConfig tableConfig = setupTableConfig(propertyStore);
- Schema schema = setupSchema(propertyStore);
- HelixManager helixManager = mock(HelixManager.class);
- when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
- RealtimeTableDataManager tmgr = new RealtimeTableDataManager(null);
- tmgr.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
-
- // Create a raw segment and put it in deep store backed by local fs.
- String segName = "seg01";
- SegmentZKMetadata segmentZKMetadata =
- TableDataManagerTestUtils.makeRawSegment(segName, createSegment(tableConfig, schema, segName),
- new File(TEMP_DIR, segName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION), true);
- segmentZKMetadata.setStatus(Status.DONE);
-
- // Local segment dir doesn't exist, thus downloading from deep store.
- File localSegDir = new File(TABLE_DATA_DIR, segName);
- assertFalse(localSegDir.exists());
- IndexLoadingConfig indexLoadingConfig =
- TableDataManagerTestUtils.createIndexLoadingConfig("default", tableConfig, schema);
- tmgr.addSegment(segName, indexLoadingConfig, segmentZKMetadata);
- // Segment data is put on default location.
- assertTrue(localSegDir.exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(new File(TABLE_DATA_DIR, segName));
- assertEquals(llmd.getTotalDocs(), 5);
- }
-
- @Test
- public void testAddSegmentDefaultTierByTierBasedDirLoader()
- throws Exception {
- InstanceDataManagerConfig instanceDataManagerConfig = createInstanceDataManagerConfig();
- ZkHelixPropertyStore<ZNRecord> propertyStore = mock(ZkHelixPropertyStore.class);
- TableConfig tableConfig = setupTableConfig(propertyStore);
- Schema schema = setupSchema(propertyStore);
- HelixManager helixManager = mock(HelixManager.class);
- when(helixManager.getHelixPropertyStore()).thenReturn(propertyStore);
- RealtimeTableDataManager tmgr1 = new RealtimeTableDataManager(null);
- tmgr1.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
-
- // Create a raw segment and put it in deep store backed by local fs.
- String segName = "seg_tiered_01";
- SegmentZKMetadata segmentZKMetadata =
- TableDataManagerTestUtils.makeRawSegment(segName, createSegment(tableConfig, schema, segName),
- new File(TEMP_DIR, segName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION), true);
- segmentZKMetadata.setStatus(Status.DONE);
-
- // Local segment dir doesn't exist, thus downloading from deep store.
- File localSegDir = new File(TABLE_DATA_DIR, segName);
- assertFalse(localSegDir.exists());
-
- // Add segment
- IndexLoadingConfig indexLoadingConfig =
- TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, schema);
- tmgr1.addSegment(segName, indexLoadingConfig, segmentZKMetadata);
- assertTrue(localSegDir.exists());
- SegmentMetadataImpl llmd = new SegmentMetadataImpl(new File(TABLE_DATA_DIR, segName));
- assertEquals(llmd.getTotalDocs(), 5);
-
- // Now, repeat initialization of the table data manager
- tmgr1.shutDown();
- RealtimeTableDataManager tmgr2 = new RealtimeTableDataManager(null);
- tmgr2.init(instanceDataManagerConfig, tableConfig, helixManager, null, null);
-
- // Reinitialize index loading config and try adding the segment
- indexLoadingConfig = TableDataManagerTestUtils.createIndexLoadingConfig("tierBased", tableConfig, schema);
- tmgr2.addSegment(segName, indexLoadingConfig, segmentZKMetadata);
-
- // Make sure that the segment hasn't been moved
- assertTrue(localSegDir.exists());
- llmd = new SegmentMetadataImpl(new File(TABLE_DATA_DIR, segName));
- assertEquals(llmd.getTotalDocs(), 5);
- }
-
- @Test
- public void testAllowDownload() {
- RealtimeTableDataManager mgr = new RealtimeTableDataManager(null);
-
- LLCSegmentName llc = new LLCSegmentName("tbl01", 0, 1000000, System.currentTimeMillis());
- SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
- when(zkmd.getStatus()).thenReturn(Status.IN_PROGRESS);
- assertFalse(mgr.allowDownload(llc.getSegmentName(), zkmd));
-
- when(zkmd.getStatus()).thenReturn(Status.DONE);
- when(zkmd.getDownloadUrl()).thenReturn("");
- assertFalse(mgr.allowDownload(llc.getSegmentName(), zkmd));
-
- when(zkmd.getDownloadUrl()).thenReturn("remote");
- assertTrue(mgr.allowDownload(llc.getSegmentName(), zkmd));
- }
-
- private static File createSegment(TableConfig tableConfig, Schema schema, String segName)
- throws Exception {
- SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, schema);
- config.setOutDir(TABLE_DATA_DIR.getAbsolutePath());
- config.setSegmentName(segName);
- config.setSegmentVersion(SegmentVersion.v3);
- List<GenericRow> rows = new ArrayList<>(3);
- for (int i = 0; i < STRING_VALUES.length; i++) {
- GenericRow row = new GenericRow();
- row.putValue(STRING_COLUMN, STRING_VALUES[i]);
- row.putValue(LONG_COLUMN, LONG_VALUES[i]);
- rows.add(row);
- }
- SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
- driver.init(config, new GenericRowRecordReader(rows));
- driver.build();
- return new File(TABLE_DATA_DIR, segName);
- }
-
- private static InstanceDataManagerConfig createInstanceDataManagerConfig() {
- InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
- when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
- return instanceDataManagerConfig;
- }
-
- private static TableConfig setupTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore)
- throws Exception {
- TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).build();
- ZNRecord tableConfigZNRecord = TableConfigUtils.toZNRecord(tableConfig);
- when(propertyStore.get(ZKMetadataProvider.constructPropertyStorePathForResourceConfig(TABLE_NAME_WITH_TYPE), null,
- AccessOption.PERSISTENT)).thenReturn(tableConfigZNRecord);
- return tableConfig;
- }
-
- private static Schema setupSchema(ZkHelixPropertyStore<ZNRecord> propertyStore) {
- Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
- .addSingleValueDimension(STRING_COLUMN, FieldSpec.DataType.STRING)
- .addMetric(LONG_COLUMN, FieldSpec.DataType.LONG).build();
- ZNRecord schemaZNRecord = SchemaUtils.toZNRecord(schema);
- when(propertyStore.get(ZKMetadataProvider.constructPropertyStorePathForSchema(TABLE_NAME), null,
- AccessOption.PERSISTENT)).thenReturn(schemaZNRecord);
- return schema;
- }
@Test
public void testSetDefaultTimeValueIfInvalid() {
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
index f066b1618b..6d932489e9 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
@@ -144,6 +144,8 @@ public class MetadataAndDictionaryAggregationPlanMakerTest {
@AfterClass
public void destroySegment() {
_indexSegment.destroy();
+ _upsertIndexSegment.offload();
+ _upsertIndexSegment.destroy();
}
@AfterTest
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/QueryOverrideWithHintsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/QueryOverrideWithHintsTest.java
index 522f588191..7b9cb63f0f 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/QueryOverrideWithHintsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/QueryOverrideWithHintsTest.java
@@ -105,6 +105,10 @@ public class QueryOverrideWithHintsTest {
return null;
}
+ @Override
+ public void offload() {
+ }
+
@Override
public void destroy() {
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorExceptionsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorExceptionsTest.java
index 49697b6b85..c664c2f013 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorExceptionsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorExceptionsTest.java
@@ -41,6 +41,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
@@ -133,8 +134,8 @@ public class QueryExecutorExceptionsTest {
InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(INDEX_DIR.getAbsolutePath());
TableDataManager tableDataManager =
- new TableDataManagerProvider(instanceDataManagerConfig).getTableDataManager(tableConfig,
- mock(HelixManager.class));
+ new TableDataManagerProvider(instanceDataManagerConfig, mock(HelixManager.class),
+ new SegmentLocks()).getTableDataManager(tableConfig);
tableDataManager.start();
//we don't add index segments to the data manager to simulate numSegmentsAcquired < numSegmentsQueried
InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
index c600d8b5cc..d59b6df681 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
@@ -38,6 +38,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
@@ -131,8 +132,8 @@ public class QueryExecutorTest {
InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManager tableDataManager =
- new TableDataManagerProvider(instanceDataManagerConfig).getTableDataManager(tableConfig,
- mock(HelixManager.class));
+ new TableDataManagerProvider(instanceDataManagerConfig, mock(HelixManager.class),
+ new SegmentLocks()).getTableDataManager(tableConfig);
tableDataManager.start();
for (ImmutableSegment indexSegment : _indexSegments) {
tableDataManager.addSegment(indexSegment);
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
index 17c543c0f7..f38c408eea 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
@@ -55,6 +55,7 @@ import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoa
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
@@ -292,8 +293,8 @@ public class ExplainPlanQueriesTest extends BaseQueriesTest {
InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManager tableDataManager =
- new TableDataManagerProvider(instanceDataManagerConfig).getTableDataManager(TABLE_CONFIG,
- mock(HelixManager.class));
+ new TableDataManagerProvider(instanceDataManagerConfig, mock(HelixManager.class),
+ new SegmentLocks()).getTableDataManager(TABLE_CONFIG);
tableDataManager.start();
for (IndexSegment indexSegment : _indexSegments) {
tableDataManager.addSegment((ImmutableSegment) indexSegment);
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/SegmentWithNullValueVectorTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/SegmentWithNullValueVectorTest.java
index ac9b7e82a9..3ef16168a2 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/SegmentWithNullValueVectorTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/SegmentWithNullValueVectorTest.java
@@ -46,6 +46,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
@@ -138,8 +139,8 @@ public class SegmentWithNullValueVectorTest {
InstanceDataManagerConfig instanceDataManagerConfig = mock(InstanceDataManagerConfig.class);
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManager tableDataManager =
- new TableDataManagerProvider(instanceDataManagerConfig).getTableDataManager(tableConfig,
- mock(HelixManager.class));
+ new TableDataManagerProvider(instanceDataManagerConfig, mock(HelixManager.class),
+ new SegmentLocks()).getTableDataManager(tableConfig);
tableDataManager.start();
tableDataManager.addSegment(_segment);
_instanceDataManager = mock(InstanceDataManager.class);
diff --git a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
index a503c6f966..6b08aec206 100644
--- a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
+++ b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
@@ -182,7 +182,7 @@ public class MockInstanceDataManagerFactory {
try (RecordReader recordReader = new GenericRowRecordReader(rows)) {
driver.init(config, recordReader);
driver.build();
- return ImmutableSegmentLoader.load(new File(indexDir, segmentName), ReadMode.mmap, tableConfig, schema);
+ return ImmutableSegmentLoader.load(new File(indexDir, segmentName), ReadMode.mmap);
} catch (Exception e) {
throw new RuntimeException("Unable to construct immutable segment from records", e);
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/SegmentDataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/SegmentDataManager.java
index 2e1dae8d1f..77b01a97e8 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/SegmentDataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/SegmentDataManager.java
@@ -28,6 +28,7 @@ import org.apache.pinot.segment.spi.IndexSegment;
*/
public abstract class SegmentDataManager {
private final long _loadTimeMs = System.currentTimeMillis();
+ private final AtomicBoolean _offloaded = new AtomicBoolean();
private final AtomicBoolean _destroyed = new AtomicBoolean();
private int _referenceCount = 1;
@@ -73,11 +74,27 @@ public abstract class SegmentDataManager {
public abstract IndexSegment getSegment();
+ /**
+ * Offloads the segment from the metadata management (e.g. upsert metadata), but not releases the resources yet
+ * because there might be queries still accessing the segment.
+ */
+ public void offload() {
+ if (_offloaded.compareAndSet(false, true)) {
+ doOffload();
+ }
+ }
+
+ public abstract void doOffload();
+
/**
* Destroys the data manager and releases all the resources allocated.
* The data manager can only be destroyed once.
*/
public void destroy() {
+ // NOTE: We want the test to catch the case when destroy is called without offloading, but not fail the production.
+ assert _offloaded.get() : "Cannot destroy segment data manager without offloading it first";
+ offload();
+
if (_destroyed.compareAndSet(false, true)) {
doDestroy();
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
index 30dbdffbf0..9f1352ec66 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
@@ -18,11 +18,13 @@
*/
package org.apache.pinot.segment.local.data.manager;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.LoadingCache;
import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.Lock;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.lang3.tuple.Pair;
@@ -30,6 +32,7 @@ import org.apache.helix.HelixManager;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.restlet.resources.SegmentErrorInfo;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentContext;
@@ -48,8 +51,8 @@ public interface TableDataManager {
/**
* Initializes the table data manager. Should be called only once and before calling any other method.
*/
- void init(InstanceDataManagerConfig instanceDataManagerConfig, TableConfig tableConfig, HelixManager helixManager,
- @Nullable ExecutorService segmentPreloadExecutor,
+ void init(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager, SegmentLocks segmentLocks,
+ TableConfig tableConfig, @Nullable ExecutorService segmentPreloadExecutor,
@Nullable LoadingCache<Pair<String, String>, SegmentErrorInfo> errorCache);
/**
@@ -76,22 +79,53 @@ public interface TableDataManager {
boolean isShutDown();
+ /**
+ * Returns the segment lock for a segment in the table.
+ */
+ Lock getSegmentLock(String segmentName);
+
+ /**
+ * Returns whether the segment is loaded in the table.
+ */
+ boolean hasSegment(String segmentName);
+
/**
* Adds a loaded immutable segment into the table.
+ * NOTE: This method is not designed to be directly used by the production code, but can be handy to set up tests.
*/
+ @VisibleForTesting
void addSegment(ImmutableSegment immutableSegment);
/**
- * Adds a segment from local disk into the OFFLINE table.
+ * Adds an ONLINE segment into a table.
+ * This method is triggered by state transition to ONLINE state.
+ * If the segment is already loaded:
+ * - If the segment is a consuming segment, it will be transitioned to immutable segment (consuming segment commit).
+ * - If the segment is already an immutable segment, replace it if its CRC doesn't match the ZK metadata.
+ */
+ void addOnlineSegment(String segmentName)
+ throws Exception;
+
+ /**
+ * Adds an CONSUMING segment into a REALTIME table.
+ * This method is triggered by state transition to CONSUMING state.
+ */
+ void addConsumingSegment(String segmentName)
+ throws Exception;
+
+ /**
+ * Replaces an already loaded segment in a table if the segment has been overridden in the deep store (CRC mismatch).
+ * This method is triggered by a custom message (NOT state transition), and the target segment should be in ONLINE
+ * state.
*/
- void addSegment(File indexDir, IndexLoadingConfig indexLoadingConfig)
+ void replaceSegment(String segmentName)
throws Exception;
/**
- * Adds a segment into the REALTIME table.
- * <p>The segment could be committed or under consuming.
+ * Offloads a segment from table but not dropping its data from server.
+ * This method is triggered by state transition to OFFLINE state.
*/
- void addSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, SegmentZKMetadata zkMetadata)
+ void offloadSegment(String segmentName)
throws Exception;
/**
@@ -115,29 +149,6 @@ public interface TableDataManager {
SegmentMetadata localMetadata, @Nullable Schema schema, boolean forceDownload)
throws Exception;
- /**
- * Adds or replaces an immutable segment for the table, which can be an OFFLINE or REALTIME table.
- * A new segment may be downloaded if the local one has a different CRC or doesn't work as expected.
- * This operation is conducted outside the failure handling framework as used in segment reloading,
- * because the segment is not yet online serving queries, e.g. this method is used to add a new segment,
- * or transition a segment to online serving state.
- *
- * @param segmentName the segment to add or replace
- * @param indexLoadingConfig the latest table config to load segment
- * @param zkMetadata the segment metadata from zookeeper
- * @param localMetadata the segment metadata object held by server, which can be null when
- * the server is restarted or the segment is newly added to the table
- * @throws Exception thrown upon failure when to add or replace the segment
- */
- void addOrReplaceSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, SegmentZKMetadata zkMetadata,
- @Nullable SegmentMetadata localMetadata)
- throws Exception;
-
- /**
- * Removes a segment from the table.
- */
- void removeSegment(String segmentName);
-
/**
* Try to load a segment from an existing segment directory managed by the server. The segment loading may fail
* because the directory may not exist any more, or the segment data has a different crc now, or the existing segment
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/EmptyIndexSegment.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/EmptyIndexSegment.java
index 681e93a0e0..c4374829c3 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/EmptyIndexSegment.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/EmptyIndexSegment.java
@@ -76,6 +76,10 @@ public class EmptyIndexSegment implements ImmutableSegment {
return _segmentMetadata.getSchema().getPhysicalColumnNames();
}
+ @Override
+ public void offload() {
+ }
+
@Override
public void destroy() {
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
index 88707df9bc..ff49a36c8d 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
@@ -252,18 +252,19 @@ public class ImmutableSegmentImpl implements ImmutableSegment {
}
@Override
- public void destroy() {
- String segmentName = getSegmentName();
- LOGGER.info("Trying to destroy segment : {}", segmentName);
-
- // Remove the upsert and dedup metadata before closing the readers
+ public void offload() {
if (_partitionUpsertMetadataManager != null) {
_partitionUpsertMetadataManager.removeSegment(this);
}
-
if (_partitionDedupMetadataManager != null) {
_partitionDedupMetadataManager.removeSegment(this);
}
+ }
+
+ @Override
+ public void destroy() {
+ String segmentName = getSegmentName();
+ LOGGER.info("Trying to destroy segment : {}", segmentName);
// StarTreeIndexContainer refers to other column index containers, so close it firstly.
if (_starTreeIndexContainer != null) {
try {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
index 546e88dda8..9bb5e84a47 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java
@@ -68,35 +68,45 @@ public class ImmutableSegmentLoader {
throws Exception {
IndexLoadingConfig defaultIndexLoadingConfig = new IndexLoadingConfig();
defaultIndexLoadingConfig.setReadMode(readMode);
- return load(indexDir, defaultIndexLoadingConfig, null, false);
+ return load(indexDir, defaultIndexLoadingConfig, false);
}
/**
* Loads the segment with specified table config and schema.
* This method is used to access the segment without modifying it, i.e. in read-only mode.
*/
+ @Deprecated
public static ImmutableSegment load(File indexDir, ReadMode readMode, TableConfig tableConfig,
@Nullable Schema schema)
throws Exception {
IndexLoadingConfig defaultIndexLoadingConfig = new IndexLoadingConfig(tableConfig, schema);
defaultIndexLoadingConfig.setReadMode(readMode);
- return load(indexDir, defaultIndexLoadingConfig, schema, false);
+ return load(indexDir, defaultIndexLoadingConfig, false);
}
/**
- * Loads the segment with empty schema but a specified IndexLoadingConfig.
+ * Loads the segment with specified IndexLoadingConfig.
* This method modifies the segment like to convert segment format, add or remove indices.
* Mostly used by UT cases to add some specific index for testing purpose.
*/
public static ImmutableSegment load(File indexDir, IndexLoadingConfig indexLoadingConfig)
throws Exception {
- return load(indexDir, indexLoadingConfig, null, true);
+ return load(indexDir, indexLoadingConfig, true);
+ }
+
+ /**
+ * Loads the segment with specified IndexLoadingConfig.
+ * This method modifies the segment like to convert segment format, add or remove indices.
+ */
+ public static ImmutableSegment load(File indexDir, IndexLoadingConfig indexLoadingConfig, boolean needPreprocess)
+ throws Exception {
+ return load(indexDir, indexLoadingConfig, indexLoadingConfig.getSchema(), needPreprocess);
}
/**
* Loads the segment with specified schema and IndexLoadingConfig, usually from Zookeeper.
* This method modifies the segment like to convert segment format, add or remove indices.
- * Mainly used during segment reloading.
+ * Mostly used by UT cases to add some specific index for testing purpose.
*/
public static ImmutableSegment load(File indexDir, IndexLoadingConfig indexLoadingConfig, @Nullable Schema schema)
throws Exception {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index 38bd1921c6..61c7ea360a 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -953,17 +953,18 @@ public class MutableSegmentImpl implements MutableSegment {
}
@Override
- public void destroy() {
- _logger.info("Trying to close RealtimeSegmentImpl : {}", _segmentName);
-
- // Remove the upsert and dedup metadata before closing the readers
+ public void offload() {
if (_partitionUpsertMetadataManager != null) {
_partitionUpsertMetadataManager.removeSegment(this);
}
-
if (_partitionDedupMetadataManager != null) {
_partitionDedupMetadataManager.removeSegment(this);
}
+ }
+
+ @Override
+ public void destroy() {
+ _logger.info("Trying to close RealtimeSegmentImpl : {}", _segmentName);
// Gather statistics for off-heap mode
if (_offHeap) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
index f8bacbf6ba..d777a9d8ff 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
@@ -62,7 +62,6 @@ import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImp
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.HashUtils;
-import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.MutableSegment;
@@ -524,20 +523,13 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps
@VisibleForTesting
public void addSegment(ImmutableSegmentImpl segment, @Nullable ThreadSafeMutableRoaringBitmap validDocIds,
@Nullable ThreadSafeMutableRoaringBitmap queryableDocIds, Iterator<RecordInfo> recordInfoIterator) {
- String segmentName = segment.getSegmentName();
- Lock segmentLock = SegmentLocks.getSegmentLock(_tableNameWithType, segmentName);
- segmentLock.lock();
- try {
- if (validDocIds == null) {
- validDocIds = new ThreadSafeMutableRoaringBitmap();
- }
- if (queryableDocIds == null && _deleteRecordColumn != null) {
- queryableDocIds = new ThreadSafeMutableRoaringBitmap();
- }
- addOrReplaceSegment(segment, validDocIds, queryableDocIds, recordInfoIterator, null, null);
- } finally {
- segmentLock.unlock();
+ if (validDocIds == null) {
+ validDocIds = new ThreadSafeMutableRoaringBitmap();
+ }
+ if (queryableDocIds == null && _deleteRecordColumn != null) {
+ queryableDocIds = new ThreadSafeMutableRoaringBitmap();
}
+ addOrReplaceSegment(segment, validDocIds, queryableDocIds, recordInfoIterator, null, null);
}
protected abstract void addOrReplaceSegment(ImmutableSegmentImpl segment, ThreadSafeMutableRoaringBitmap validDocIds,
@@ -637,51 +629,43 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps
public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutableRoaringBitmap validDocIds,
@Nullable ThreadSafeMutableRoaringBitmap queryableDocIds, @Nullable Iterator<RecordInfo> recordInfoIterator,
IndexSegment oldSegment) {
- String segmentName = segment.getSegmentName();
- Lock segmentLock = SegmentLocks.getSegmentLock(_tableNameWithType, segmentName);
-
// Currently when TTL is enabled, we don't support skip loading out-of-TTL segment with snapshots, since we don't
// know which docs are valid in the new segment.
// TODO: when ttl is enabled, we can allow
// (1) skip loading segments without any invalid docs.
// (2) assign the invalid docs from the replaced segment to the new segment.
- segmentLock.lock();
- try {
- MutableRoaringBitmap validDocIdsForOldSegment =
- oldSegment.getValidDocIds() != null ? oldSegment.getValidDocIds().getMutableRoaringBitmap() : null;
- if (recordInfoIterator != null) {
- Preconditions.checkArgument(segment instanceof ImmutableSegmentImpl,
- "Got unsupported segment implementation: {} for segment: {}, table: {}", segment.getClass(), segmentName,
- _tableNameWithType);
- if (validDocIds == null) {
- validDocIds = new ThreadSafeMutableRoaringBitmap();
- }
- if (queryableDocIds == null && _deleteRecordColumn != null) {
- queryableDocIds = new ThreadSafeMutableRoaringBitmap();
- }
- addOrReplaceSegment((ImmutableSegmentImpl) segment, validDocIds, queryableDocIds, recordInfoIterator,
- oldSegment, validDocIdsForOldSegment);
+ String segmentName = segment.getSegmentName();
+ MutableRoaringBitmap validDocIdsForOldSegment =
+ oldSegment.getValidDocIds() != null ? oldSegment.getValidDocIds().getMutableRoaringBitmap() : null;
+ if (recordInfoIterator != null) {
+ Preconditions.checkArgument(segment instanceof ImmutableSegmentImpl,
+ "Got unsupported segment implementation: {} for segment: {}, table: {}", segment.getClass(), segmentName,
+ _tableNameWithType);
+ if (validDocIds == null) {
+ validDocIds = new ThreadSafeMutableRoaringBitmap();
}
-
- if (validDocIdsForOldSegment != null && !validDocIdsForOldSegment.isEmpty()) {
- int numKeysNotReplaced = validDocIdsForOldSegment.getCardinality();
- if (_partialUpsertHandler != null) {
- // For partial-upsert table, because we do not restore the original record location when removing the primary
- // keys not replaced, it can potentially cause inconsistency between replicas. This can happen when a
- // consuming segment is replaced by a committed segment that is consumed from a different server with
- // different records (some stream consumer cannot guarantee consuming the messages in the same order).
- _logger.warn("Found {} primary keys not replaced when replacing segment: {} for partial-upsert table. This "
- + "can potentially cause inconsistency between replicas", numKeysNotReplaced, segmentName);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.PARTIAL_UPSERT_KEYS_NOT_REPLACED,
- numKeysNotReplaced);
- } else {
- _logger.info("Found {} primary keys not replaced when replacing segment: {}", numKeysNotReplaced,
- segmentName);
- }
- removeSegment(oldSegment, validDocIdsForOldSegment);
+ if (queryableDocIds == null && _deleteRecordColumn != null) {
+ queryableDocIds = new ThreadSafeMutableRoaringBitmap();
}
- } finally {
- segmentLock.unlock();
+ addOrReplaceSegment((ImmutableSegmentImpl) segment, validDocIds, queryableDocIds, recordInfoIterator, oldSegment,
+ validDocIdsForOldSegment);
+ }
+
+ if (validDocIdsForOldSegment != null && !validDocIdsForOldSegment.isEmpty()) {
+ int numKeysNotReplaced = validDocIdsForOldSegment.getCardinality();
+ if (_partialUpsertHandler != null) {
+ // For partial-upsert table, because we do not restore the original record location when removing the primary
+ // keys not replaced, it can potentially cause inconsistency between replicas. This can happen when a
+ // consuming segment is replaced by a committed segment that is consumed from a different server with
+ // different records (some stream consumer cannot guarantee consuming the messages in the same order).
+ _logger.warn("Found {} primary keys not replaced when replacing segment: {} for partial-upsert table. This "
+ + "can potentially cause inconsistency between replicas", numKeysNotReplaced, segmentName);
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.PARTIAL_UPSERT_KEYS_NOT_REPLACED,
+ numKeysNotReplaced);
+ } else {
+ _logger.info("Found {} primary keys not replaced when replacing segment: {}", numKeysNotReplaced, segmentName);
+ }
+ removeSegment(oldSegment, validDocIdsForOldSegment);
}
}
@@ -739,14 +723,8 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps
return;
}
- Lock segmentLock = SegmentLocks.getSegmentLock(_tableNameWithType, segmentName);
- segmentLock.lock();
- try {
- _logger.info("Removing {} primary keys for segment: {}", validDocIds.getCardinality(), segmentName);
- removeSegment(segment, validDocIds);
- } finally {
- segmentLock.unlock();
- }
+ _logger.info("Removing {} primary keys for segment: {}", validDocIds.getCardinality(), segmentName);
+ removeSegment(segment, validDocIds);
// Update metrics
long numPrimaryKeys = getNumPrimaryKeys();
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
index dc34e4bdec..1c72470242 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
@@ -67,6 +67,8 @@ public interface PartitionUpsertMetadataManager extends Closeable {
/**
* Preload segments for the table partition. Segments can be added differently during preloading.
+ * TODO: Revisit this and see if we can use the same IndexLoadingConfig for all segments. Tier info might be different
+ * for different segments.
*/
void preloadSegments(IndexLoadingConfig indexLoadingConfig);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentLocks.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentLocks.java
index a242b19956..1a2cbbd230 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentLocks.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentLocks.java
@@ -23,12 +23,12 @@ import java.util.concurrent.locks.ReentrantLock;
public class SegmentLocks {
- private static final SegmentLocks DEFAULT_LOCKS = create();
private static final int DEFAULT_NUM_LOCKS = 10000;
+
private final Lock[] _locks;
private final int _numLocks;
- private SegmentLocks(int numLocks) {
+ public SegmentLocks(int numLocks) {
_numLocks = numLocks;
_locks = new Lock[numLocks];
for (int i = 0; i < numLocks; i++) {
@@ -36,18 +36,30 @@ public class SegmentLocks {
}
}
+ public SegmentLocks() {
+ this(DEFAULT_NUM_LOCKS);
+ }
+
public Lock getLock(String tableNameWithType, String segmentName) {
return _locks[Math.abs((31 * tableNameWithType.hashCode() + segmentName.hashCode()) % _numLocks)];
}
+ // DO NOT use global lock because that can break tests with multiple server instances
+
+ @Deprecated
+ private static final SegmentLocks DEFAULT_LOCKS = create();
+
+ @Deprecated
public static Lock getSegmentLock(String tableNameWithType, String segmentName) {
return DEFAULT_LOCKS.getLock(tableNameWithType, segmentName);
}
+ @Deprecated
public static SegmentLocks create() {
return new SegmentLocks(DEFAULT_NUM_LOCKS);
}
+ @Deprecated
public static SegmentLocks create(int numLocks) {
return new SegmentLocks(numLocks);
}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
index 9c48a00efc..53f7b99558 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
@@ -130,6 +130,12 @@ public interface IndexSegment {
default void release(FetchContext fetchContext) {
}
+ /**
+ * Offloads the segment from the metadata management (e.g. upsert metadata), but not releases the resources yet
+ * because there might be queries still accessing the segment.
+ */
+ void offload();
+
/**
* Destroys segment in memory and closes file handlers if in MMAP mode.
*/
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
index f097c4103c..ed5e087b41 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
@@ -84,6 +84,8 @@ public class HelixInstanceDataManager implements InstanceDataManager {
private static final Logger LOGGER = LoggerFactory.getLogger(HelixInstanceDataManager.class);
private final ConcurrentHashMap<String, TableDataManager> _tableDataManagerMap = new ConcurrentHashMap<>();
+ // TODO: Consider making segment locks per table instead of per instance
+ private final SegmentLocks _segmentLocks = new SegmentLocks();
private HelixInstanceDataManagerConfig _instanceDataManagerConfig;
private String _instanceId;
@@ -114,8 +116,8 @@ public class HelixInstanceDataManager implements InstanceDataManager {
_instanceDataManagerConfig = new HelixInstanceDataManagerConfig(config);
LOGGER.info("HelixInstanceDataManagerConfig: {}", _instanceDataManagerConfig);
_instanceId = _instanceDataManagerConfig.getInstanceId();
- _tableDataManagerProvider = new TableDataManagerProvider(_instanceDataManagerConfig);
_helixManager = helixManager;
+ _tableDataManagerProvider = new TableDataManagerProvider(_instanceDataManagerConfig, helixManager, _segmentLocks);
_segmentUploader = new PinotFSSegmentUploader(_instanceDataManagerConfig.getSegmentStoreUri(),
ServerSegmentCompletionProtocolHandler.getSegmentUploadRequestTimeoutMs(), serverMetrics);
@@ -152,9 +154,9 @@ public class HelixInstanceDataManager implements InstanceDataManager {
// Initialize the error cache
_errorCache = CacheBuilder.newBuilder().maximumSize(_instanceDataManagerConfig.getErrorCacheSize())
- .build(new CacheLoader<Pair<String, String>, SegmentErrorInfo>() {
+ .build(new CacheLoader<>() {
@Override
- public SegmentErrorInfo load(Pair<String, String> tableNameWithTypeSegmentNamePair) {
+ public SegmentErrorInfo load(Pair<String, String> tableSegmentPair) {
// This cache is populated only via the put api.
return null;
}
@@ -225,34 +227,6 @@ public class HelixInstanceDataManager implements InstanceDataManager {
LOGGER.info("Helix instance data manager shut down");
}
- @Override
- public void addRealtimeSegment(String realtimeTableName, String segmentName)
- throws Exception {
- LOGGER.info("Adding segment: {} to table: {}", segmentName, realtimeTableName);
- TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, realtimeTableName);
- Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", realtimeTableName);
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, tableConfig);
- Preconditions.checkState(schema != null, "Failed to find schema for table: %s", realtimeTableName);
- SegmentZKMetadata zkMetadata =
- ZKMetadataProvider.getSegmentZKMetadata(_propertyStore, realtimeTableName, segmentName);
- Preconditions.checkState(zkMetadata != null, "Failed to find ZK metadata for segment: %s, table: %s", segmentName,
- realtimeTableName);
- _tableDataManagerMap.computeIfAbsent(realtimeTableName, k -> createTableDataManager(tableConfig))
- .addSegment(segmentName, new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema), zkMetadata);
- LOGGER.info("Added segment: {} to table: {}", segmentName, realtimeTableName);
- }
-
- private TableDataManager createTableDataManager(TableConfig tableConfig) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Creating table data manager for table: {}", tableNameWithType);
- TableDataManager tableDataManager =
- _tableDataManagerProvider.getTableDataManager(tableConfig, _helixManager, _segmentPreloadExecutor, _errorCache,
- _isServerReadyToServeQueries);
- tableDataManager.start();
- LOGGER.info("Created table data manager for table: {}", tableNameWithType);
- return tableDataManager;
- }
-
@Override
public void deleteTable(String tableNameWithType)
throws Exception {
@@ -296,14 +270,52 @@ public class HelixInstanceDataManager implements InstanceDataManager {
}
@Override
- public void offloadSegment(String tableNameWithType, String segmentName) {
- LOGGER.info("Removing segment: {} from table: {}", segmentName, tableNameWithType);
+ public void addOnlineSegment(String tableNameWithType, String segmentName)
+ throws Exception {
+ _tableDataManagerMap.computeIfAbsent(tableNameWithType, this::createTableDataManager).addOnlineSegment(segmentName);
+ }
+
+ @Override
+ public void addConsumingSegment(String realtimeTableName, String segmentName)
+ throws Exception {
+ _tableDataManagerMap.computeIfAbsent(realtimeTableName, this::createTableDataManager)
+ .addConsumingSegment(segmentName);
+ }
+
+ private TableDataManager createTableDataManager(String tableNameWithType) {
+ LOGGER.info("Creating table data manager for table: {}", tableNameWithType);
+ TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", tableNameWithType);
+ TableDataManager tableDataManager =
+ _tableDataManagerProvider.getTableDataManager(tableConfig, _segmentPreloadExecutor, _errorCache,
+ _isServerReadyToServeQueries);
+ tableDataManager.start();
+ LOGGER.info("Created table data manager for table: {}", tableNameWithType);
+ return tableDataManager;
+ }
+
+ @Override
+ public void replaceSegment(String tableNameWithType, String segmentName)
+ throws Exception {
+ LOGGER.info("Replacing segment: {} in table: {}", segmentName, tableNameWithType);
+ TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
+ if (tableDataManager != null) {
+ tableDataManager.replaceSegment(segmentName);
+ } else {
+ LOGGER.warn("Failed to find data manager for table: {}, skipping replacing segment: {}", tableNameWithType,
+ segmentName);
+ }
+ }
+
+ @Override
+ public void offloadSegment(String tableNameWithType, String segmentName)
+ throws Exception {
+ LOGGER.info("Offloading segment: {} from table: {}", segmentName, tableNameWithType);
TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
if (tableDataManager != null) {
- tableDataManager.removeSegment(segmentName);
- LOGGER.info("Removed segment: {} from table: {}", segmentName, tableNameWithType);
+ tableDataManager.offloadSegment(segmentName);
} else {
- LOGGER.warn("Failed to find data manager for table: {}, skipping removing segment: {}", tableNameWithType,
+ LOGGER.warn("Failed to find data manager for table: {}, skipping offloading segment: {}", tableNameWithType,
segmentName);
}
}
@@ -311,19 +323,29 @@ public class HelixInstanceDataManager implements InstanceDataManager {
@Override
public void deleteSegment(String tableNameWithType, String segmentName)
throws Exception {
- // This method might modify the file on disk. Use segment lock to prevent race condition
- Lock segmentLock = SegmentLocks.getSegmentLock(tableNameWithType, segmentName);
+ LOGGER.info("Deleting segment: {} from table: {}", segmentName, tableNameWithType);
+ // Segment deletion is handled at instance level because table data manager might not exist. Acquire the lock here.
+ Lock segmentLock = _segmentLocks.getLock(tableNameWithType, segmentName);
+ segmentLock.lock();
try {
- segmentLock.lock();
-
+ // Check if the segment is still loaded, if so, offload it first.
+ // This might happen when the server disconnected from ZK and reconnected, and the segment is still loaded.
+ // TODO: Consider using table data manager to delete the segment. This will allow the table data manager to clean
+ // up the segment data on all tiers. Note that table data manager might have not been created, and table
+ // config might have been deleted at this point.
+ TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
+ if (tableDataManager != null && tableDataManager.hasSegment(segmentName)) {
+ LOGGER.warn("Segment: {} from table: {} is still loaded, offloading it first", segmentName, tableNameWithType);
+ tableDataManager.offloadSegment(segmentName);
+ }
// Clean up the segment data on default tier unconditionally.
File segmentDir = getSegmentDataDirectory(tableNameWithType, segmentName);
if (segmentDir.exists()) {
FileUtils.deleteQuietly(segmentDir);
LOGGER.info("Deleted segment directory {} on default tier", segmentDir);
}
- // We might clean up further more with the specific segment loader. But note that tableDataManager object or
- // even the TableConfig might not be present any more at this point.
+ // We might clean up further more with the specific segment loader. But note that table data manager might have
+ // not been created, and table config might have been deleted at this point.
SegmentDirectoryLoader segmentLoader = SegmentDirectoryLoaderRegistry.getSegmentDirectoryLoader(
_instanceDataManagerConfig.getSegmentDirectoryLoader());
if (segmentLoader != null) {
@@ -333,11 +355,13 @@ public class HelixInstanceDataManager implements InstanceDataManager {
.setTableDataDir(_instanceDataManagerConfig.getInstanceDataDir() + "/" + tableNameWithType).build();
segmentLoader.delete(ctx);
}
+ LOGGER.info("Deleted segment: {} from table: {}", segmentName, tableNameWithType);
} finally {
segmentLock.unlock();
}
}
+ // TODO: Move reload handling logic to table data manager
@Override
public void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload)
throws Exception {
@@ -358,6 +382,7 @@ public class HelixInstanceDataManager implements InstanceDataManager {
LOGGER.info("Reloaded single segment: {} in table: {}", segmentName, tableNameWithType);
}
+ // TODO: Move reload handling logic to table data manager
@Override
public void reloadAllSegments(String tableNameWithType, boolean forceDownload,
SegmentRefreshSemaphore segmentRefreshSemaphore)
@@ -367,6 +392,7 @@ public class HelixInstanceDataManager implements InstanceDataManager {
reloadSegmentsWithMetadata(tableNameWithType, segmentsMetadata, forceDownload, segmentRefreshSemaphore);
}
+ // TODO: Move reload handling logic to table data manager
@Override
public void reloadSegments(String tableNameWithType, List<String> segmentNames, boolean forceDownload,
SegmentRefreshSemaphore segmentRefreshSemaphore)
@@ -470,56 +496,13 @@ public class HelixInstanceDataManager implements InstanceDataManager {
}
}
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
+ indexLoadingConfig.setErrorOnColumnBuildFailure(true);
SegmentZKMetadata zkMetadata =
ZKMetadataProvider.getSegmentZKMetadata(_propertyStore, tableNameWithType, segmentName);
- Preconditions.checkNotNull(zkMetadata);
-
- // This method might modify the file on disk. Use segment lock to prevent race condition
- Lock segmentLock = SegmentLocks.getSegmentLock(tableNameWithType, segmentName);
- try {
- segmentLock.lock();
-
- // Reloads an existing segment, and the local segment metadata is existing as asserted above.
- IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
- indexLoadingConfig.setErrorOnColumnBuildFailure(true);
- tableDataManager.reloadSegment(segmentName, indexLoadingConfig, zkMetadata, segmentMetadata, schema,
- forceDownload);
- LOGGER.info("Reloaded segment: {} of table: {}", segmentName, tableNameWithType);
- } finally {
- segmentLock.unlock();
- }
- }
-
- @Override
- public void addOrReplaceSegment(String tableNameWithType, String segmentName)
- throws Exception {
- LOGGER.info("Adding or replacing segment: {} for table: {}", segmentName, tableNameWithType);
-
- // Get updated table config, schema and segment metadata from Zookeeper.
- TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType);
- Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", tableNameWithType);
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, tableConfig);
- SegmentZKMetadata zkMetadata =
- ZKMetadataProvider.getSegmentZKMetadata(_propertyStore, tableNameWithType, segmentName);
- Preconditions.checkState(zkMetadata != null, "Failed to find ZK metadata for segment: %s, table: %s", segmentName,
+ Preconditions.checkState(zkMetadata != null, "Failed to find ZK metadata for segment: %s of table: %s", segmentName,
tableNameWithType);
-
- // This method might modify the file on disk. Use segment lock to prevent race condition
- Lock segmentLock = SegmentLocks.getSegmentLock(tableNameWithType, segmentName);
- try {
- segmentLock.lock();
-
- // But if table mgr is not created or the segment is not loaded yet, the localMetadata
- // is set to null. Then, addOrReplaceSegment method will load the segment accordingly.
- SegmentMetadata localMetadata = getSegmentMetadata(tableNameWithType, segmentName);
-
- _tableDataManagerMap.computeIfAbsent(tableNameWithType, k -> createTableDataManager(tableConfig))
- .addOrReplaceSegment(segmentName, new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema),
- zkMetadata, localMetadata);
- LOGGER.info("Added or replaced segment: {} of table: {}", segmentName, tableNameWithType);
- } finally {
- segmentLock.unlock();
- }
+ tableDataManager.reloadSegment(segmentName, indexLoadingConfig, zkMetadata, segmentMetadata, schema, forceDownload);
}
@Override
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
index 7fe52f472b..5e4f54140f 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
@@ -101,10 +101,10 @@ public class SegmentMessageHandlerFactory implements MessageHandlerFactory {
throws InterruptedException {
HelixTaskResult result = new HelixTaskResult();
_logger.info("Handling message: {}", _message);
+ _segmentRefreshSemaphore.acquireSema(_segmentName, _logger);
try {
- _segmentRefreshSemaphore.acquireSema(_segmentName, _logger);
// The number of retry times depends on the retry count in Constants.
- _instanceDataManager.addOrReplaceSegment(_tableNameWithType, _segmentName);
+ _instanceDataManager.replaceSegment(_tableNameWithType, _segmentName);
result.setSuccess(true);
} catch (Exception e) {
_metrics.addMeteredTableValue(_tableNameWithType, ServerMeter.REFRESH_FAILURES, 1);
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentOnlineOfflineStateModelFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentOnlineOfflineStateModelFactory.java
index b0625c879b..daa2790205 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentOnlineOfflineStateModelFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentOnlineOfflineStateModelFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.server.starter.helix;
-import com.google.common.base.Preconditions;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import java.util.concurrent.TimeUnit;
@@ -29,15 +28,8 @@ import org.apache.helix.participant.statemachine.StateModel;
import org.apache.helix.participant.statemachine.StateModelFactory;
import org.apache.helix.participant.statemachine.StateModelInfo;
import org.apache.helix.participant.statemachine.Transition;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.restlet.resources.SegmentErrorInfo;
import org.apache.pinot.core.data.manager.InstanceDataManager;
-import org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager;
-import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
-import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -83,95 +75,38 @@ public class SegmentOnlineOfflineStateModelFactory extends StateModelFactory<Sta
private final Logger _logger = LoggerFactory.getLogger(_instanceId + " - SegmentOnlineOfflineStateModel");
@Transition(from = "OFFLINE", to = "CONSUMING")
- public void onBecomeConsumingFromOffline(Message message, NotificationContext context) {
+ public void onBecomeConsumingFromOffline(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeConsumingFromOffline() : " + message);
- String realtimeTableName = message.getResourceName();
- String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.addRealtimeSegment(realtimeTableName, segmentName);
- } catch (Exception e) {
- String errorMessage =
- String.format("Caught exception in state transition OFFLINE -> CONSUMING for table: %s, segment: %s",
- realtimeTableName, segmentName);
- _logger.error(errorMessage, e);
- TableDataManager tableDataManager = _instanceDataManager.getTableDataManager(realtimeTableName);
- if (tableDataManager != null) {
- tableDataManager.addSegmentError(segmentName,
- new SegmentErrorInfo(System.currentTimeMillis(), errorMessage, e));
- }
- Utils.rethrowException(e);
- }
+ _instanceDataManager.addConsumingSegment(message.getResourceName(), message.getPartitionName());
}
@Transition(from = "CONSUMING", to = "ONLINE")
- public void onBecomeOnlineFromConsuming(Message message, NotificationContext context) {
+ public void onBecomeOnlineFromConsuming(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeOnlineFromConsuming() : " + message);
- String realtimeTableName = message.getResourceName();
- String segmentName = message.getPartitionName();
- TableDataManager tableDataManager = _instanceDataManager.getTableDataManager(realtimeTableName);
- Preconditions.checkState(tableDataManager != null, "Failed to find table: %s", realtimeTableName);
- tableDataManager.onConsumingToOnline(segmentName);
- SegmentDataManager acquiredSegment = tableDataManager.acquireSegment(segmentName);
- // For this transition to be correct in helix, we should already have a segment that is consuming
- Preconditions.checkState(acquiredSegment != null, "Failed to find segment: %s in table: %s", segmentName,
- realtimeTableName);
-
- // TODO: https://github.com/apache/pinot/issues/10049
- try {
- if (!(acquiredSegment instanceof RealtimeSegmentDataManager)) {
- // We found an LLC segment that is not consuming right now, must be that we already swapped it with a
- // segment that has been built. Nothing to do for this state transition.
- _logger.info("Segment {} not an instance of RealtimeSegmentDataManager. Reporting success for the transition",
- acquiredSegment.getSegmentName());
- return;
- }
- RealtimeSegmentDataManager segmentDataManager = (RealtimeSegmentDataManager) acquiredSegment;
- SegmentZKMetadata segmentZKMetadata =
- ZKMetadataProvider.getSegmentZKMetadata(_instanceDataManager.getPropertyStore(), realtimeTableName,
- segmentName);
- segmentDataManager.goOnlineFromConsuming(segmentZKMetadata);
- } catch (Exception e) {
- String errorMessage =
- String.format("Caught exception in state transition CONSUMING -> ONLINE for table: %s, segment: %s",
- realtimeTableName, segmentName);
- _logger.error(errorMessage, e);
- tableDataManager.addSegmentError(segmentName,
- new SegmentErrorInfo(System.currentTimeMillis(), errorMessage, e));
- Utils.rethrowException(e);
- } finally {
- tableDataManager.releaseSegment(acquiredSegment);
- }
+ _instanceDataManager.addOnlineSegment(message.getResourceName(), message.getPartitionName());
}
@Transition(from = "CONSUMING", to = "OFFLINE")
- public void onBecomeOfflineFromConsuming(Message message, NotificationContext context) {
+ public void onBecomeOfflineFromConsuming(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeOfflineFromConsuming() : " + message);
String realtimeTableName = message.getResourceName();
String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.offloadSegment(realtimeTableName, segmentName);
- _recentlyOffloadedConsumingSegments.put(Pair.of(realtimeTableName, segmentName), true);
- } catch (Exception e) {
- _logger.error("Caught exception in state transition CONSUMING -> OFFLINE for table: {}, segment: {}",
- realtimeTableName, segmentName, e);
- Utils.rethrowException(e);
- }
+ _instanceDataManager.offloadSegment(realtimeTableName, segmentName);
+ _recentlyOffloadedConsumingSegments.put(Pair.of(realtimeTableName, segmentName), true);
}
@Transition(from = "CONSUMING", to = "DROPPED")
- public void onBecomeDroppedFromConsuming(Message message, NotificationContext context) {
+ public void onBecomeDroppedFromConsuming(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeDroppedFromConsuming() : " + message);
String realtimeTableName = message.getResourceName();
String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.offloadSegment(realtimeTableName, segmentName);
- _instanceDataManager.deleteSegment(realtimeTableName, segmentName);
- onConsumingToDropped(realtimeTableName, segmentName);
- } catch (Exception e) {
- _logger.error("Caught exception in state transition CONSUMING -> DROPPED for table: {}, segment: {}",
- realtimeTableName, segmentName, e);
- Utils.rethrowException(e);
- }
+ _instanceDataManager.offloadSegment(realtimeTableName, segmentName);
+ _instanceDataManager.deleteSegment(realtimeTableName, segmentName);
+ onConsumingToDropped(realtimeTableName, segmentName);
}
/**
@@ -190,85 +125,45 @@ public class SegmentOnlineOfflineStateModelFactory extends StateModelFactory<Sta
}
@Transition(from = "OFFLINE", to = "ONLINE")
- public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+ public void onBecomeOnlineFromOffline(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeOnlineFromOffline() : " + message);
- String tableNameWithType = message.getResourceName();
- String segmentName = message.getPartitionName();
- try {
- TableType tableType = TableNameBuilder.getTableTypeFromTableName(tableNameWithType);
- Preconditions.checkNotNull(tableType);
- if (tableType == TableType.OFFLINE) {
- _instanceDataManager.addOrReplaceSegment(tableNameWithType, segmentName);
- } else {
- _instanceDataManager.addRealtimeSegment(tableNameWithType, segmentName);
- }
- } catch (Exception e) {
- String errorMessage =
- String.format("Caught exception in state transition OFFLINE -> ONLINE for table: %s, segment: %s",
- tableNameWithType, segmentName);
- _logger.error(errorMessage, e);
- TableDataManager tableDataManager = _instanceDataManager.getTableDataManager(tableNameWithType);
- if (tableDataManager != null) {
- tableDataManager.addSegmentError(segmentName,
- new SegmentErrorInfo(System.currentTimeMillis(), errorMessage, e));
- }
- Utils.rethrowException(e);
- }
+ _instanceDataManager.addOnlineSegment(message.getResourceName(), message.getPartitionName());
}
- // Remove segment from InstanceDataManager.
- // Still keep the data files in local.
@Transition(from = "ONLINE", to = "OFFLINE")
- public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+ public void onBecomeOfflineFromOnline(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeOfflineFromOnline() : " + message);
- String tableNameWithType = message.getResourceName();
- String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.offloadSegment(tableNameWithType, segmentName);
- } catch (Exception e) {
- _logger.error("Caught exception in state transition ONLINE -> OFFLINE for table: {}, segment: {}",
- tableNameWithType, segmentName, e);
- Utils.rethrowException(e);
- }
+ _instanceDataManager.offloadSegment(message.getResourceName(), message.getPartitionName());
}
- // Delete segment from local directory.
@Transition(from = "OFFLINE", to = "DROPPED")
- public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
+ public void onBecomeDroppedFromOffline(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeDroppedFromOffline() : " + message);
String tableNameWithType = message.getResourceName();
String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.deleteSegment(tableNameWithType, segmentName);
-
- // Check if the segment is recently offloaded from CONSUMING to OFFLINE
- if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
- Pair<String, String> tableSegmentPair = Pair.of(tableNameWithType, segmentName);
- if (_recentlyOffloadedConsumingSegments.getIfPresent(tableSegmentPair) != null) {
- _recentlyOffloadedConsumingSegments.invalidate(tableSegmentPair);
- onConsumingToDropped(tableNameWithType, segmentName);
- }
+ _instanceDataManager.deleteSegment(tableNameWithType, segmentName);
+
+ // Check if the segment is recently offloaded from CONSUMING to OFFLINE
+ if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+ Pair<String, String> tableSegmentPair = Pair.of(tableNameWithType, segmentName);
+ if (_recentlyOffloadedConsumingSegments.getIfPresent(tableSegmentPair) != null) {
+ _recentlyOffloadedConsumingSegments.invalidate(tableSegmentPair);
+ onConsumingToDropped(tableNameWithType, segmentName);
}
- } catch (Exception e) {
- _logger.error("Caught exception in state transition OFFLINE -> DROPPED for table: {}, segment: {}",
- tableNameWithType, segmentName, e);
- Utils.rethrowException(e);
}
}
@Transition(from = "ONLINE", to = "DROPPED")
- public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
+ public void onBecomeDroppedFromOnline(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeDroppedFromOnline() : " + message);
String tableNameWithType = message.getResourceName();
String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.offloadSegment(tableNameWithType, segmentName);
- _instanceDataManager.deleteSegment(tableNameWithType, segmentName);
- } catch (Exception e) {
- _logger.error("Caught exception in state transition ONLINE -> DROPPED for table: {}, segment: {}",
- tableNameWithType, segmentName, e);
- Utils.rethrowException(e);
- }
+ _instanceDataManager.offloadSegment(tableNameWithType, segmentName);
+ _instanceDataManager.deleteSegment(tableNameWithType, segmentName);
}
@Transition(from = "ERROR", to = "OFFLINE")
@@ -277,17 +172,10 @@ public class SegmentOnlineOfflineStateModelFactory extends StateModelFactory<Sta
}
@Transition(from = "ERROR", to = "DROPPED")
- public void onBecomeDroppedFromError(Message message, NotificationContext context) {
+ public void onBecomeDroppedFromError(Message message, NotificationContext context)
+ throws Exception {
_logger.info("SegmentOnlineOfflineStateModel.onBecomeDroppedFromError() : " + message);
- String tableNameWithType = message.getResourceName();
- String segmentName = message.getPartitionName();
- try {
- _instanceDataManager.deleteSegment(tableNameWithType, segmentName);
- } catch (Exception e) {
- _logger.error("Caught exception in state transition ERROR -> DROPPED for table: {}, segment: {}",
- tableNameWithType, segmentName, e);
- Utils.rethrowException(e);
- }
+ _instanceDataManager.deleteSegment(message.getResourceName(), message.getPartitionName());
}
}
}
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java b/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
index e57371f72d..f274053db3 100644
--- a/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
+++ b/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
@@ -42,6 +42,7 @@ import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.SegmentTestUtils;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
@@ -152,9 +153,11 @@ public abstract class BaseResourceTest {
public void tearDown() {
_adminApiApplication.stop();
for (ImmutableSegment immutableSegment : _realtimeIndexSegments) {
+ immutableSegment.offload();
immutableSegment.destroy();
}
for (ImmutableSegment immutableSegment : _offlineIndexSegments) {
+ immutableSegment.offload();
immutableSegment.destroy();
}
FileUtils.deleteQuietly(TEMP_DIR);
@@ -198,7 +201,8 @@ public abstract class BaseResourceTest {
// NOTE: Use OfflineTableDataManager for both OFFLINE and REALTIME table because RealtimeTableDataManager requires
// table config.
TableDataManager tableDataManager = new OfflineTableDataManager();
- tableDataManager.init(instanceDataManagerConfig, tableConfig, mock(HelixManager.class), null, null);
+ tableDataManager.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), tableConfig, null,
+ null);
tableDataManager.start();
_tableDataManagerMap.put(tableNameWithType, tableDataManager);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org