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 2025/04/03 05:21:41 UTC
(pinot) branch table_data_manager_cache_config created (now 078808af4c)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a change to branch table_data_manager_cache_config
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 078808af4c Cache TableConfig and Schema in TableDataManager
This branch includes the following new commits:
new 078808af4c Cache TableConfig and Schema in TableDataManager
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Cache TableConfig and Schema in TableDataManager
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch table_data_manager_cache_config
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 078808af4c92d417004684b44a90566d2b645991
Author: Xiaotian (Jackie) Jiang <ja...@gmail.com>
AuthorDate: Wed Apr 2 18:48:49 2025 -0700
Cache TableConfig and Schema in TableDataManager
---
.../core/data/manager/BaseTableDataManager.java | 44 +++++++++++++++-------
.../manager/offline/DimensionTableDataManager.java | 22 ++++-------
.../provider/DefaultTableDataManagerProvider.java | 10 +++--
.../manager/provider/TableDataManagerProvider.java | 12 +++---
.../manager/realtime/RealtimeTableDataManager.java | 15 ++++----
.../BaseTableDataManagerAcquireSegmentTest.java | 10 +++--
.../data/manager/BaseTableDataManagerTest.java | 3 +-
.../offline/DimensionTableDataManagerTest.java | 12 +++---
.../realtime/RealtimeSegmentDataManagerTest.java | 3 +-
.../executor/QueryExecutorExceptionsTest.java | 2 +-
.../core/query/executor/QueryExecutorTest.java | 2 +-
.../pinot/queries/ExplainPlanQueriesTest.java | 2 +-
.../queries/SegmentWithNullValueVectorTest.java | 2 +-
.../FailureInjectingTableDataManagerProvider.java | 11 +++---
.../perf/BenchmarkDimensionTableOverhead.java | 2 +-
.../local/data/manager/TableDataManager.java | 26 ++++++-------
.../pinot/server/api/resources/TablesResource.java | 5 +--
.../starter/helix/HelixInstanceDataManager.java | 7 +++-
.../apache/pinot/server/api/BaseResourceTest.java | 3 +-
19 files changed, 105 insertions(+), 88 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 c506dc336c..254b2eb58f 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
@@ -103,6 +103,7 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.TimestampIndexUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -119,7 +120,6 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected HelixManager _helixManager;
protected ZkHelixPropertyStore<ZNRecord> _propertyStore;
protected SegmentLocks _segmentLocks;
- protected TableConfig _tableConfig;
protected String _tableNameWithType;
protected String _tableDataDir;
protected File _indexDir;
@@ -136,6 +136,12 @@ public abstract class BaseTableDataManager implements TableDataManager {
protected boolean _isStreamSegmentDownloadUntar;
@Nullable
protected SegmentOperationsThrottler _segmentOperationsThrottler;
+
+ // Caches the latest TableConfig and Schema. The cached TableConfig and Schema already have timestamp index applied,
+ // and should not be modified.
+ protected volatile TableConfig _tableConfig;
+ protected volatile Schema _schema;
+
// Semaphore to restrict the maximum number of parallel segment downloads from deep store for a table
private Semaphore _segmentDownloadSemaphore;
private AtomicInteger _numSegmentsAcquiredDownloadSemaphore;
@@ -150,7 +156,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
@Override
public void init(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager,
- SegmentLocks segmentLocks, TableConfig tableConfig, SegmentReloadSemaphore segmentReloadSemaphore,
+ SegmentLocks segmentLocks, TableConfig tableConfig, Schema schema, SegmentReloadSemaphore segmentReloadSemaphore,
ExecutorService segmentReloadExecutor, @Nullable ExecutorService segmentPreloadExecutor,
@Nullable Cache<Pair<String, String>, SegmentErrorInfo> errorCache,
@Nullable SegmentOperationsThrottler segmentOperationsThrottler) {
@@ -161,14 +167,16 @@ public abstract class BaseTableDataManager implements TableDataManager {
_helixManager = helixManager;
_propertyStore = helixManager.getHelixPropertyStore();
_segmentLocks = segmentLocks;
+ TimestampIndexUtils.applyTimestampIndex(tableConfig, schema);
_tableConfig = tableConfig;
+ _schema = schema;
_segmentReloadSemaphore = segmentReloadSemaphore;
_segmentReloadExecutor = segmentReloadExecutor;
_segmentPreloadExecutor = segmentPreloadExecutor;
- _authProvider = AuthProviderUtils.extractAuthProvider(_instanceDataManagerConfig.getAuthConfig(), null);
+ _authProvider = AuthProviderUtils.extractAuthProvider(instanceDataManagerConfig.getAuthConfig(), null);
_tableNameWithType = tableConfig.getTableName();
- _tableDataDir = _instanceDataManagerConfig.getInstanceDataDir() + File.separator + _tableNameWithType;
+ _tableDataDir = instanceDataManagerConfig.getInstanceDataDir() + File.separator + _tableNameWithType;
_indexDir = new File(_tableDataDir);
if (!_indexDir.exists()) {
Preconditions.checkState(_indexDir.mkdirs(), "Unable to create index directory at %s. "
@@ -379,21 +387,29 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
@Override
- public Pair<TableConfig, Schema> fetchTableConfigAndSchema() {
+ public IndexLoadingConfig fetchIndexLoadingConfig() {
TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, _tableNameWithType);
Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", _tableNameWithType);
Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
Preconditions.checkState(schema != null, "Failed to find schema for table: %s", _tableNameWithType);
- return Pair.of(tableConfig, schema);
- }
-
- @Override
- public IndexLoadingConfig getIndexLoadingConfig(TableConfig tableConfig, Schema schema) {
IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
indexLoadingConfig.setTableDataDir(_tableDataDir);
+ // NOTE: Timestamp index is already applied in the constructor of IndexLoadingConfig.
+ _tableConfig = tableConfig;
+ _schema = schema;
return indexLoadingConfig;
}
+ @Override
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ @Override
+ public Schema getSchema() {
+ return _schema;
+ }
+
@Override
public void addNewOnlineSegment(SegmentZKMetadata zkMetadata, IndexLoadingConfig indexLoadingConfig)
throws Exception {
@@ -1258,10 +1274,12 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
@Override
- public List<StaleSegment> getStaleSegments(TableConfig tableConfig, Schema schema) {
+ public List<StaleSegment> getStaleSegments() {
List<StaleSegment> staleSegments = new ArrayList<>();
List<SegmentDataManager> segmentDataManagers = acquireAllSegments();
- final long startTime = System.currentTimeMillis();
+ TableConfig tableConfig = _tableConfig;
+ Schema schema = _schema;
+ long startTimeMs = System.currentTimeMillis();
try {
for (SegmentDataManager segmentDataManager : segmentDataManagers) {
StaleSegment response = isSegmentStale(tableConfig, schema, segmentDataManager);
@@ -1273,7 +1291,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
for (SegmentDataManager segmentDataManager : segmentDataManagers) {
releaseSegment(segmentDataManager);
}
- LOGGER.info("Time Taken to get stale segments: {} ms", System.currentTimeMillis() - startTime);
+ LOGGER.info("Time Taken to get stale segments: {} ms", System.currentTimeMillis() - startTimeMs);
}
return staleSegments;
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 5e36d847e4..0c0a7240ab 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
@@ -35,14 +35,12 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.core.data.manager.provider.TableDataManagerProvider;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.spi.config.table.DimensionTableConfig;
-import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -108,20 +106,16 @@ public class DimensionTableDataManager extends OfflineTableDataManager {
@Override
protected void doInit() {
super.doInit();
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
- Preconditions.checkState(schema != null, "Failed to find schema for dimension table: %s", _tableNameWithType);
+ Schema schema = _schema;
List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
Preconditions.checkState(CollectionUtils.isNotEmpty(primaryKeyColumns),
"Primary key columns must be configured for dimension table: %s", _tableNameWithType);
- TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, _tableNameWithType);
- if (tableConfig != null) {
- DimensionTableConfig dimensionTableConfig = tableConfig.getDimensionTableConfig();
- if (dimensionTableConfig != null) {
- _disablePreload = dimensionTableConfig.isDisablePreload();
- _errorOnDuplicatePrimaryKey = dimensionTableConfig.isErrorOnDuplicatePrimaryKey();
- }
+ DimensionTableConfig dimensionTableConfig = _tableConfig.getDimensionTableConfig();
+ if (dimensionTableConfig != null) {
+ _disablePreload = dimensionTableConfig.isDisablePreload();
+ _errorOnDuplicatePrimaryKey = dimensionTableConfig.isErrorOnDuplicatePrimaryKey();
}
if (_disablePreload) {
@@ -206,8 +200,7 @@ public class DimensionTableDataManager extends OfflineTableDataManager {
// loading is in progress.
int token = _loadToken.incrementAndGet();
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
- Preconditions.checkState(schema != null, "Failed to find schema for dimension table: %s", _tableNameWithType);
+ Schema schema = _schema;
List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
Preconditions.checkState(CollectionUtils.isNotEmpty(primaryKeyColumns),
"Primary key columns must be configured for dimension table: %s", _tableNameWithType);
@@ -282,8 +275,7 @@ public class DimensionTableDataManager extends OfflineTableDataManager {
// loading is in progress.
int token = _loadToken.incrementAndGet();
- Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
- Preconditions.checkState(schema != null, "Failed to find schema for dimension table: %s", _tableNameWithType);
+ Schema schema = _schema;
List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
Preconditions.checkState(CollectionUtils.isNotEmpty(primaryKeyColumns),
"Primary key columns must be configured for dimension table: %s", _tableNameWithType);
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/DefaultTableDataManagerProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/DefaultTableDataManagerProvider.java
index 186736002d..b750016f74 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/DefaultTableDataManagerProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/DefaultTableDataManagerProvider.java
@@ -37,6 +37,7 @@ import org.apache.pinot.segment.local.utils.SegmentOperationsThrottler;
import org.apache.pinot.segment.local.utils.SegmentReloadSemaphore;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
@@ -64,8 +65,9 @@ public class DefaultTableDataManagerProvider implements TableDataManagerProvider
}
@Override
- public TableDataManager getTableDataManager(TableConfig tableConfig, SegmentReloadSemaphore segmentReloadSemaphore,
- ExecutorService segmentReloadExecutor, @Nullable ExecutorService segmentPreloadExecutor,
+ public TableDataManager getTableDataManager(TableConfig tableConfig, Schema schema,
+ SegmentReloadSemaphore segmentReloadSemaphore, ExecutorService segmentReloadExecutor,
+ @Nullable ExecutorService segmentPreloadExecutor,
@Nullable Cache<Pair<String, String>, SegmentErrorInfo> errorCache,
Supplier<Boolean> isServerReadyToServeQueries) {
TableDataManager tableDataManager;
@@ -90,8 +92,8 @@ public class DefaultTableDataManagerProvider implements TableDataManagerProvider
default:
throw new IllegalStateException();
}
- tableDataManager.init(_instanceDataManagerConfig, _helixManager, _segmentLocks, tableConfig, segmentReloadSemaphore,
- segmentReloadExecutor, segmentPreloadExecutor, errorCache, _segmentOperationsThrottler);
+ tableDataManager.init(_instanceDataManagerConfig, _helixManager, _segmentLocks, tableConfig, schema,
+ segmentReloadSemaphore, segmentReloadExecutor, segmentPreloadExecutor, errorCache, _segmentOperationsThrottler);
return tableDataManager;
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/TableDataManagerProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/TableDataManagerProvider.java
index 59e5a75f8c..63bfed0b0e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/TableDataManagerProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/provider/TableDataManagerProvider.java
@@ -34,6 +34,7 @@ import org.apache.pinot.segment.local.utils.SegmentReloadSemaphore;
import org.apache.pinot.spi.annotations.InterfaceAudience;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
/**
@@ -45,14 +46,15 @@ public interface TableDataManagerProvider {
void init(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager, SegmentLocks segmentLocks,
@Nullable SegmentOperationsThrottler segmentOperationsThrottler);
- TableDataManager getTableDataManager(TableConfig tableConfig, SegmentReloadSemaphore segmentRefreshSemaphore,
- ExecutorService segmentRefreshExecutor, @Nullable ExecutorService segmentPreloadExecutor,
+ TableDataManager getTableDataManager(TableConfig tableConfig, Schema schema,
+ SegmentReloadSemaphore segmentRefreshSemaphore, ExecutorService segmentRefreshExecutor,
+ @Nullable ExecutorService segmentPreloadExecutor,
@Nullable Cache<Pair<String, String>, SegmentErrorInfo> errorCache,
Supplier<Boolean> isServerReadyToServeQueries);
@VisibleForTesting
- default TableDataManager getTableDataManager(TableConfig tableConfig) {
- return getTableDataManager(tableConfig, new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null,
- null, () -> true);
+ default TableDataManager getTableDataManager(TableConfig tableConfig, Schema schema) {
+ return getTableDataManager(tableConfig, schema, new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(),
+ null, null, () -> true);
}
}
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 5d212530b2..9c82d43fca 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
@@ -201,7 +201,8 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
// Set up dedup/upsert metadata manager
// NOTE: Dedup/upsert has to be set up when starting the server. Changing the table config without restarting the
// server won't enable/disable them on the fly.
- DedupConfig dedupConfig = _tableConfig.getDedupConfig();
+ TableConfig tableConfig = _tableConfig;
+ DedupConfig dedupConfig = tableConfig.getDedupConfig();
boolean dedupEnabled = dedupConfig != null && dedupConfig.isDedupEnabled();
if (dedupEnabled) {
Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
@@ -210,19 +211,19 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
Preconditions.checkState(!CollectionUtils.isEmpty(primaryKeyColumns),
"Primary key columns must be configured for dedup");
- _tableDedupMetadataManager = TableDedupMetadataManagerFactory.create(_tableConfig, schema, this, _serverMetrics,
+ _tableDedupMetadataManager = TableDedupMetadataManagerFactory.create(tableConfig, schema, this, _serverMetrics,
_instanceDataManagerConfig.getDedupConfig());
}
- UpsertConfig upsertConfig = _tableConfig.getUpsertConfig();
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
if (upsertConfig != null && upsertConfig.getMode() != UpsertConfig.Mode.NONE) {
Preconditions.checkState(!dedupEnabled, "Dedup and upsert cannot be both enabled for table: %s",
_tableUpsertMetadataManager);
Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _tableNameWithType);
Preconditions.checkState(schema != null, "Failed to find schema for table: %s", _tableNameWithType);
_tableUpsertMetadataManager =
- TableUpsertMetadataManagerFactory.create(_tableConfig, _instanceDataManagerConfig.getUpsertConfig());
- _tableUpsertMetadataManager.init(_tableConfig, schema, this);
+ TableUpsertMetadataManagerFactory.create(tableConfig, _instanceDataManagerConfig.getUpsertConfig());
+ _tableUpsertMetadataManager.init(tableConfig, schema, this);
}
_enforceConsumptionInOrder = isEnforceConsumptionInOrder();
@@ -594,9 +595,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
String segmentName = zkMetadata.getSegmentName();
Preconditions.checkState(status == Status.COMMITTING, "Invalid status: %s for segment: %s to be downloaded", status,
segmentName);
- TableConfig tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, _tableNameWithType);
- Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", _tableNameWithType);
- long downloadTimeoutMs = getDownloadTimeoutMs(tableConfig);
+ long downloadTimeoutMs = getDownloadTimeoutMs(_tableConfig);
long deadlineMs = System.currentTimeMillis() + downloadTimeoutMs;
while (System.currentTimeMillis() < deadlineMs) {
// ZK Metadata may change during segment download process; fetch it on every retry.
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 ad43c7c299..499c3e14a8 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
@@ -47,6 +47,7 @@ import org.apache.pinot.segment.spi.SegmentMetadata;
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.builder.TableConfigBuilder;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
@@ -123,12 +124,13 @@ public class BaseTableDataManagerAcquireSegmentTest {
when(instanceDataManagerConfig.getDeletedSegmentsCacheSize()).thenReturn(DELETED_SEGMENTS_CACHE_SIZE);
when(instanceDataManagerConfig.getDeletedSegmentsCacheTtlMinutes()).thenReturn(DELETED_SEGMENTS_TTL_MINUTES);
TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
- SegmentOperationsThrottler segmentOperationsThrottler = new SegmentOperationsThrottler(
- new SegmentAllIndexPreprocessThrottler(8, 10, true), new SegmentStarTreePreprocessThrottler(4, 8, true),
- new SegmentDownloadThrottler(10, 20, true));
+ SegmentOperationsThrottler segmentOperationsThrottler =
+ new SegmentOperationsThrottler(new SegmentAllIndexPreprocessThrottler(8, 10, true),
+ new SegmentStarTreePreprocessThrottler(4, 8, true), new SegmentDownloadThrottler(10, 20, true));
TableDataManager tableDataManager = new OfflineTableDataManager();
tableDataManager.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), tableConfig,
- new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null, segmentOperationsThrottler);
+ mock(Schema.class), new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null,
+ segmentOperationsThrottler);
tableDataManager.start();
Field segsMapField = BaseTableDataManager.class.getDeclaredField("_segmentDataManagerMap");
segsMapField.setAccessible(true);
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 c992c800da..163cd8a0c5 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
@@ -664,7 +664,8 @@ public class BaseTableDataManagerTest {
private static OfflineTableDataManager createTableManager(InstanceDataManagerConfig instanceDataManagerConfig) {
OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
tableDataManager.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), DEFAULT_TABLE_CONFIG,
- new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null, SEGMENT_OPERATIONS_THROTTLER);
+ SCHEMA, new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null,
+ SEGMENT_OPERATIONS_THROTTLER);
return tableDataManager;
}
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 372d5ce9c2..cd5c0c7c9b 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
@@ -89,7 +89,6 @@ public class DimensionTableDataManagerTest {
new SegmentDownloadThrottler(1, 2, true));
private File _indexDir;
- private SegmentMetadata _segmentMetadata;
private SegmentZKMetadata _segmentZKMetadata;
@BeforeClass
@@ -120,9 +119,9 @@ public class DimensionTableDataManagerTest {
String segmentName = driver.getSegmentName();
_indexDir = new File(tableDataDir, segmentName);
- _segmentMetadata = new SegmentMetadataImpl(_indexDir);
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(_indexDir);
_segmentZKMetadata = new SegmentZKMetadata(segmentName);
- _segmentZKMetadata.setCrc(Long.parseLong(_segmentMetadata.getCrc()));
+ _segmentZKMetadata.setCrc(Long.parseLong(segmentMetadata.getCrc()));
}
@AfterClass
@@ -174,7 +173,7 @@ public class DimensionTableDataManagerTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
DimensionTableDataManager tableDataManager =
DimensionTableDataManager.createInstanceByTableName(OFFLINE_TABLE_NAME);
- tableDataManager.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), tableConfig,
+ tableDataManager.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), tableConfig, schema,
new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null, SEGMENT_OPERATIONS_THROTTLER);
tableDataManager.start();
return tableDataManager;
@@ -294,8 +293,9 @@ public class DimensionTableDataManagerTest {
Schema schemaWithExtraColumn = getSchemaWithExtraColumn();
when(propertyStore.get("/SCHEMAS/dimBaseballTeams", null, AccessOption.PERSISTENT)).thenReturn(
SchemaUtils.toZNRecord(schemaWithExtraColumn));
- tableDataManager.reloadSegment(_segmentZKMetadata.getSegmentName(),
- new IndexLoadingConfig(tableConfig, schemaWithExtraColumn), _segmentZKMetadata, _segmentMetadata, false);
+ when(propertyStore.get("/SEGMENTS/dimBaseballTeams_OFFLINE/" + _segmentZKMetadata.getSegmentName(), null,
+ AccessOption.PERSISTENT)).thenReturn(_segmentZKMetadata.toZNRecord());
+ tableDataManager.reloadSegment(_segmentZKMetadata.getSegmentName(), false);
// Confirm the new column is available for lookup
teamCitySpec = tableDataManager.getColumnFieldSpec("teamCity");
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 8bf32c576b..887f8885e2 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
@@ -752,6 +752,7 @@ public class RealtimeSegmentDataManagerTest {
throws Exception {
TableConfig tableConfig = createTableConfig();
tableConfig.setUpsertConfig(null);
+ Schema schema = mock(Schema.class);
ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class);
when(propertyStore.get(anyString(), any(), anyInt())).thenReturn(TableConfigUtils.toZNRecord(tableConfig));
HelixManager helixManager = mock(HelixManager.class);
@@ -761,7 +762,7 @@ public class RealtimeSegmentDataManagerTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManagerProvider tableDataManagerProvider = new DefaultTableDataManagerProvider();
tableDataManagerProvider.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), null);
- TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig);
+ TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig, schema);
tableDataManager.start();
tableDataManager.shutDown();
Assert.assertFalse(SegmentBuildTimeLeaseExtender.isExecutorShutdown());
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 9afed0b634..7c9bd2e4f4 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
@@ -136,7 +136,7 @@ public class QueryExecutorExceptionsTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(INDEX_DIR.getAbsolutePath());
TableDataManagerProvider tableDataManagerProvider = new DefaultTableDataManagerProvider();
tableDataManagerProvider.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), null);
- TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig);
+ TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig, schema);
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 095a1c0a89..022d134d36 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
@@ -160,7 +160,7 @@ public class QueryExecutorTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManagerProvider tableDataManagerProvider = new DefaultTableDataManagerProvider();
tableDataManagerProvider.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), null);
- TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig);
+ TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig, schema);
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 aa2d9e2cc1..db1af9416c 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
@@ -279,7 +279,7 @@ public class ExplainPlanQueriesTest extends BaseQueriesTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManagerProvider tableDataManagerProvider = new DefaultTableDataManagerProvider();
tableDataManagerProvider.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), null);
- TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(TABLE_CONFIG);
+ TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(TABLE_CONFIG, SCHEMA);
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 1c509ca5e8..ffbd50a3f5 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
@@ -141,7 +141,7 @@ public class SegmentWithNullValueVectorTest {
when(instanceDataManagerConfig.getInstanceDataDir()).thenReturn(TEMP_DIR.getAbsolutePath());
TableDataManagerProvider tableDataManagerProvider = new DefaultTableDataManagerProvider();
tableDataManagerProvider.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), null);
- TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig);
+ TableDataManager tableDataManager = tableDataManagerProvider.getTableDataManager(tableConfig, _schema);
tableDataManager.start();
tableDataManager.addSegment(_segment);
_instanceDataManager = mock(InstanceDataManager.class);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/FailureInjectingTableDataManagerProvider.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/FailureInjectingTableDataManagerProvider.java
index 7fbc220806..67cca949e9 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/FailureInjectingTableDataManagerProvider.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/FailureInjectingTableDataManagerProvider.java
@@ -37,6 +37,7 @@ import org.apache.pinot.segment.local.utils.SegmentOperationsThrottler;
import org.apache.pinot.segment.local.utils.SegmentReloadSemaphore;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
@@ -62,8 +63,9 @@ public class FailureInjectingTableDataManagerProvider implements TableDataManage
}
@Override
- public TableDataManager getTableDataManager(TableConfig tableConfig, SegmentReloadSemaphore segmentReloadSemaphore,
- ExecutorService segmentReloadExecutor, @Nullable ExecutorService segmentPreloadExecutor,
+ public TableDataManager getTableDataManager(TableConfig tableConfig, Schema schema,
+ SegmentReloadSemaphore segmentReloadSemaphore, ExecutorService segmentReloadExecutor,
+ @Nullable ExecutorService segmentPreloadExecutor,
@Nullable Cache<Pair<String, String>, SegmentErrorInfo> errorCache,
Supplier<Boolean> isServerReadyToServeQueries) {
TableDataManager tableDataManager;
@@ -89,9 +91,8 @@ public class FailureInjectingTableDataManagerProvider implements TableDataManage
default:
throw new IllegalStateException();
}
- tableDataManager.init(_instanceDataManagerConfig, _helixManager, _segmentLocks, tableConfig, segmentReloadSemaphore,
- segmentReloadExecutor, segmentPreloadExecutor,
- errorCache, null);
+ tableDataManager.init(_instanceDataManagerConfig, _helixManager, _segmentLocks, tableConfig, schema,
+ segmentReloadSemaphore, segmentReloadExecutor, segmentPreloadExecutor, errorCache, null);
return tableDataManager;
}
}
diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkDimensionTableOverhead.java b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkDimensionTableOverhead.java
index 3ed62ec35a..d2aa77321d 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkDimensionTableOverhead.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkDimensionTableOverhead.java
@@ -184,7 +184,7 @@ public class BenchmarkDimensionTableOverhead extends BaseQueriesTest {
String tableName = TABLE_NAME + "_" + _iteration;
_tableDataManager = DimensionTableDataManager.createInstanceByTableName(tableName);
- _tableDataManager.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), tableConfig,
+ _tableDataManager.init(instanceDataManagerConfig, helixManager, new SegmentLocks(), tableConfig, SCHEMA,
new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null, SEGMENT_OPERATIONS_THROTTLER);
_tableDataManager.start();
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 ed1f96f4da..111c2ec4e5 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
@@ -53,8 +53,8 @@ public interface TableDataManager {
* Initializes the table data manager. Should be called only once and before calling any other method.
*/
void init(InstanceDataManagerConfig instanceDataManagerConfig, HelixManager helixManager, SegmentLocks segmentLocks,
- TableConfig tableConfig, SegmentReloadSemaphore segmentReloadSemaphore, ExecutorService segmentReloadExecutor,
- @Nullable ExecutorService segmentPreloadExecutor,
+ TableConfig tableConfig, Schema schema, SegmentReloadSemaphore segmentReloadSemaphore,
+ ExecutorService segmentReloadExecutor, @Nullable ExecutorService segmentPreloadExecutor,
@Nullable Cache<Pair<String, String>, SegmentErrorInfo> errorCache,
@Nullable SegmentOperationsThrottler segmentOperationsThrottler);
@@ -295,22 +295,21 @@ public interface TableDataManager {
SegmentZKMetadata fetchZKMetadata(String segmentName);
/**
- * Fetches the table config and schema for the table from ZK.
+ * Fetches the table config and schema for the table from ZK, then construct the index loading config with them.
*/
- Pair<TableConfig, Schema> fetchTableConfigAndSchema();
+ IndexLoadingConfig fetchIndexLoadingConfig();
/**
- * Fetches the table config and schema for the table from ZK, then construct the index loading config with them.
+ * Returns the cached latest table config (with timestamp index applied) for the table. The cache is refreshed when
+ * invoking {@link #fetchIndexLoadingConfig()}.
*/
- default IndexLoadingConfig fetchIndexLoadingConfig() {
- Pair<TableConfig, Schema> tableConfigSchemaPair = fetchTableConfigAndSchema();
- return getIndexLoadingConfig(tableConfigSchemaPair.getLeft(), tableConfigSchemaPair.getRight());
- }
+ TableConfig getTableConfig();
/**
- * Constructs the index loading config for the table with the given table config and schema.
+ * Returns the cached latest schema (with timestamp index applied) for the table. The cache is refreshed when invoking
+ * {@link #fetchIndexLoadingConfig()}.
*/
- IndexLoadingConfig getIndexLoadingConfig(TableConfig tableConfig, Schema schema);
+ Schema getSchema();
/**
* Interface to handle segment state transitions from CONSUMING to DROPPED
@@ -330,9 +329,8 @@ public interface TableDataManager {
/**
* Return list of segment names that are stale along with reason.
- * @param tableConfig Table Config of the table
- * @param schema Schema of the table
+ *
* @return List of {@link StaleSegment} with segment names and reason why it is stale
*/
- List<StaleSegment> getStaleSegments(TableConfig tableConfig, Schema schema);
+ List<StaleSegment> getStaleSegments();
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
index f7038ed11f..3df7a4290e 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
@@ -99,11 +99,9 @@ import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.server.access.AccessControlFactory;
import org.apache.pinot.server.api.AdminApiApplication;
import org.apache.pinot.server.starter.ServerInstance;
-import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
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.stream.ConsumerPartitionState;
import org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -1188,8 +1186,7 @@ public class TablesResource {
tableName = DatabaseUtils.translateTableName(tableName, headers);
TableDataManager tableDataManager = ServerResourceUtils.checkGetTableDataManager(_serverInstance, tableName);
try {
- Pair<TableConfig, Schema> tableConfigSchemaPair = tableDataManager.fetchTableConfigAndSchema();
- return tableDataManager.getStaleSegments(tableConfigSchemaPair.getLeft(), tableConfigSchemaPair.getRight());
+ return tableDataManager.getStaleSegments();
} catch (Exception e) {
throw new WebApplicationException(e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR);
}
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 b79a759e76..b3ed24fa81 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
@@ -63,6 +63,7 @@ import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderContext;
import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderRegistry;
import org.apache.pinot.server.realtime.ServerSegmentCompletionProtocolHandler;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -294,9 +295,11 @@ public class HelixInstanceDataManager implements InstanceDataManager {
tableConfig = ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType);
Preconditions.checkState(tableConfig != null, "Failed to find table config for table: %s", tableNameWithType);
}
+ Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, tableNameWithType);
+ Preconditions.checkState(schema != null, "Failed to find schema for table: %s", tableNameWithType);
TableDataManager tableDataManager =
- _tableDataManagerProvider.getTableDataManager(tableConfig, _segmentReloadSemaphore, _segmentReloadExecutor,
- _segmentPreloadExecutor, _errorCache, _isServerReadyToServeQueries);
+ _tableDataManagerProvider.getTableDataManager(tableConfig, schema, _segmentReloadSemaphore,
+ _segmentReloadExecutor, _segmentPreloadExecutor, _errorCache, _isServerReadyToServeQueries);
tableDataManager.start();
LOGGER.info("Created table data manager for table: {}", tableNameWithType);
return tableDataManager;
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 1cc0b05b83..89cc6ef292 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
@@ -53,6 +53,7 @@ import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.NetUtils;
@@ -204,7 +205,7 @@ public abstract class BaseResourceTest {
// table config.
TableDataManager tableDataManager = new OfflineTableDataManager();
tableDataManager.init(instanceDataManagerConfig, mock(HelixManager.class), new SegmentLocks(), tableConfig,
- new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, null, null);
+ mock(Schema.class), new SegmentReloadSemaphore(1), Executors.newSingleThreadExecutor(), null, 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