Copilot commented on code in PR #19254:
URL: https://github.com/apache/pinot/pull/19254#discussion_r3800183823
##########
pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java:
##########
@@ -890,4 +894,149 @@ public static boolean
isTableConfigExists(ZkHelixPropertyStore<ZNRecord> propert
return
propertyStore.exists(constructPropertyStorePathForResourceConfig(tableNameWithType),
AccessOption.PERSISTENT);
}
+
+ /// Creates a deletion marker for a table to indicate that deletion is in
progress.
+ /// This marker is used to prevent concurrent deletions and prevent
recreation of a table
+ /// while deletion is in progress.
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @param controllerId The ID of the controller performing the deletion
+ /// @return true if the marker was created successfully, false if a marker
already exists
+ public static boolean
createTableDeletionMarker(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType, String controllerId) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ // create() is atomic: with concurrent callers only one succeeds, the rest
get false.
+ return propertyStore.create(markerPath,
buildTableDeletionMarkerRecord(tableNameWithType, controllerId),
+ AccessOption.PERSISTENT);
+ }
+
+ private static ZNRecord buildTableDeletionMarkerRecord(String
tableNameWithType, String controllerId) {
+ ZNRecord markerRecord = new ZNRecord(tableNameWithType);
+ markerRecord.setSimpleField(DELETION_MARKER_CONTROLLER_ID_KEY,
controllerId);
+ markerRecord.setLongField(DELETION_MARKER_START_TIME_KEY,
System.currentTimeMillis());
+ return markerRecord;
+ }
+
+ /// Returns true when the marker record is present and has not passed
[#DELETION_MARKER_EXPIRY_MS].
+ /// A marker with a missing or unparseable start time is treated as stale so
that a malformed record can never
+ /// block deletion or re-creation forever.
+ private static boolean isTableDeletionMarkerValid(@Nullable ZNRecord
markerRecord) {
+ if (markerRecord == null) {
+ return false;
+ }
+ long startTimeMs =
markerRecord.getLongField(DELETION_MARKER_START_TIME_KEY, -1L);
+ if (startTimeMs < 0) {
+ return false;
+ }
+ return System.currentTimeMillis() - startTimeMs <
DELETION_MARKER_EXPIRY_MS;
+ }
+
+ /// Checks if a deletion marker exists for the table and is still valid (not
expired).
+ /// A marker is considered expired if it is older than
DELETION_MARKER_EXPIRY_MS (24 hours).
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @return true if a valid (non-expired) deletion marker exists, false
otherwise
+ public static boolean
isValidTableDeletionMarkerExists(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ return isTableDeletionMarkerValid(propertyStore.get(markerPath, null,
AccessOption.PERSISTENT));
+ }
+
+ /// Removes the deletion marker for a table. Call this when a deletion
finishes, successfully or not.
+ ///
+ /// Releasing is a no-op unless `controllerId` still owns the marker. If a
deletion stalled past the expiry
+ /// window another controller may have taken the marker over and started its
own deletion; the stalled
+ /// controller must not then release a lock it no longer holds.
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @param controllerId The ID of the controller that expects to own the
marker
+ public static void removeTableDeletionMarker(ZkHelixPropertyStore<ZNRecord>
propertyStore,
+ String tableNameWithType, String controllerId) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ ZNRecord markerRecord = propertyStore.get(markerPath, null,
AccessOption.PERSISTENT);
+ if (markerRecord == null) {
+ return;
+ }
+ String ownerId =
markerRecord.getSimpleField(DELETION_MARKER_CONTROLLER_ID_KEY);
+ if (!Objects.equals(ownerId, controllerId)) {
+ LOGGER.warn("Not releasing the deletion marker for table: {}: it is now
owned by controller: {}, not: {}",
+ tableNameWithType, ownerId, controllerId);
+ return;
+ }
+ propertyStore.remove(markerPath, AccessOption.PERSISTENT);
Review Comment:
This ownership check and removal are not atomic. The old owner can read its
marker, another controller can version-take it over, and then this
unconditional remove deletes the new owner's marker. Also, `controllerId` is
shared by concurrent requests on one controller, so it cannot distinguish
successive acquisitions. Release should use an acquisition-unique token and a
version-checked delete/transaction.
##########
pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java:
##########
@@ -890,4 +894,149 @@ public static boolean
isTableConfigExists(ZkHelixPropertyStore<ZNRecord> propert
return
propertyStore.exists(constructPropertyStorePathForResourceConfig(tableNameWithType),
AccessOption.PERSISTENT);
}
+
+ /// Creates a deletion marker for a table to indicate that deletion is in
progress.
+ /// This marker is used to prevent concurrent deletions and prevent
recreation of a table
+ /// while deletion is in progress.
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @param controllerId The ID of the controller performing the deletion
+ /// @return true if the marker was created successfully, false if a marker
already exists
+ public static boolean
createTableDeletionMarker(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType, String controllerId) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ // create() is atomic: with concurrent callers only one succeeds, the rest
get false.
+ return propertyStore.create(markerPath,
buildTableDeletionMarkerRecord(tableNameWithType, controllerId),
+ AccessOption.PERSISTENT);
+ }
+
+ private static ZNRecord buildTableDeletionMarkerRecord(String
tableNameWithType, String controllerId) {
+ ZNRecord markerRecord = new ZNRecord(tableNameWithType);
+ markerRecord.setSimpleField(DELETION_MARKER_CONTROLLER_ID_KEY,
controllerId);
+ markerRecord.setLongField(DELETION_MARKER_START_TIME_KEY,
System.currentTimeMillis());
+ return markerRecord;
+ }
+
+ /// Returns true when the marker record is present and has not passed
[#DELETION_MARKER_EXPIRY_MS].
+ /// A marker with a missing or unparseable start time is treated as stale so
that a malformed record can never
+ /// block deletion or re-creation forever.
+ private static boolean isTableDeletionMarkerValid(@Nullable ZNRecord
markerRecord) {
+ if (markerRecord == null) {
+ return false;
+ }
+ long startTimeMs =
markerRecord.getLongField(DELETION_MARKER_START_TIME_KEY, -1L);
+ if (startTimeMs < 0) {
+ return false;
+ }
+ return System.currentTimeMillis() - startTimeMs <
DELETION_MARKER_EXPIRY_MS;
+ }
+
+ /// Checks if a deletion marker exists for the table and is still valid (not
expired).
+ /// A marker is considered expired if it is older than
DELETION_MARKER_EXPIRY_MS (24 hours).
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @return true if a valid (non-expired) deletion marker exists, false
otherwise
+ public static boolean
isValidTableDeletionMarkerExists(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ return isTableDeletionMarkerValid(propertyStore.get(markerPath, null,
AccessOption.PERSISTENT));
+ }
+
+ /// Removes the deletion marker for a table. Call this when a deletion
finishes, successfully or not.
+ ///
+ /// Releasing is a no-op unless `controllerId` still owns the marker. If a
deletion stalled past the expiry
+ /// window another controller may have taken the marker over and started its
own deletion; the stalled
+ /// controller must not then release a lock it no longer holds.
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @param controllerId The ID of the controller that expects to own the
marker
+ public static void removeTableDeletionMarker(ZkHelixPropertyStore<ZNRecord>
propertyStore,
+ String tableNameWithType, String controllerId) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ ZNRecord markerRecord = propertyStore.get(markerPath, null,
AccessOption.PERSISTENT);
+ if (markerRecord == null) {
+ return;
+ }
+ String ownerId =
markerRecord.getSimpleField(DELETION_MARKER_CONTROLLER_ID_KEY);
+ if (!Objects.equals(ownerId, controllerId)) {
+ LOGGER.warn("Not releasing the deletion marker for table: {}: it is now
owned by controller: {}, not: {}",
+ tableNameWithType, ownerId, controllerId);
+ return;
+ }
+ propertyStore.remove(markerPath, AccessOption.PERSISTENT);
+ }
+
+ /// Acquires the deletion marker, taking it over if the existing one is
stale.
+ /// This handles the case where a controller crashes during deletion,
leaving a stale marker behind.
+ ///
+ /// @param propertyStore The Helix property store
+ /// @param tableNameWithType The table name with type suffix (e.g.,
myTable_REALTIME)
+ /// @param controllerId The ID of the controller performing the deletion
+ /// @return true if the marker was created (either new or by taking over
expired), false if a valid marker exists
+ public static boolean
createOrTakeoverTableDeletionMarker(ZkHelixPropertyStore<ZNRecord>
propertyStore,
+ String tableNameWithType, String controllerId) {
+ String markerPath =
constructPropertyStorePathForTableDeletionMarker(tableNameWithType);
+ Stat stat = new Stat();
+ ZNRecord markerRecord = propertyStore.get(markerPath, stat,
AccessOption.PERSISTENT);
+ if (markerRecord == null) {
+ return createTableDeletionMarker(propertyStore, tableNameWithType,
controllerId);
+ }
+ if (isTableDeletionMarkerValid(markerRecord)) {
+ return false;
+ }
+ // The marker is stale, so reclaim it. Pass the version we just observed
so the write is rejected if another
+ // controller reclaimed it first.
+ return takeoverStaleTableDeletionMarker(propertyStore, tableNameWithType,
controllerId, stat.getVersion());
Review Comment:
The stale takeover does not actually preserve mutual exclusion: expiring
this persistent marker does not revoke or fence the delete that originally
acquired it. If that delete is paused for more than 24 hours and later resumes,
the takeover starts a second delete while the first can continue removing
metadata (including metadata from a later recreation). A version check only
chooses one takeover winner; use a session-bound lock, or a renewable
lease/fencing token that every destructive step validates before allowing
takeover.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java:
##########
@@ -324,52 +326,72 @@ public void removeSegmentsFromStoreInBatch(String
tableNameWithType, Collection<
if (retentionMs <= 0) {
filesToDelete.add(fileToDeleteURI);
} else {
- moveSegmentsToDeletedDir(segmentId, deletedSegmentsRetentionMs,
rawTableName, pinotFS, fileToDeleteURI);
+ // Collect move operations for batched processing
+ String deletedFileName = deletedSegmentsRetentionMs == null ?
URIUtils.encode(segmentId)
+ : getDeletedSegmentFileName(URIUtils.encode(segmentId),
deletedSegmentsRetentionMs);
+ URI deletedSegmentMoveDestURI = URIUtils.getUri(_dataDir,
DELETED_SEGMENTS, rawTableName, deletedFileName);
+ filesToMove.add(fileToDeleteURI);
+ moveDestinations.add(deletedSegmentMoveDestURI);
}
}
try {
+ // Batch delete segments with retention <= 0
if (!filesToDelete.isEmpty()) {
LOGGER.info("Deleting {} segment files", filesToDelete.size());
pinotFS.deleteBatch(filesToDelete, true);
}
+
+ // Move the segments to the deleted directory and let the retention
manager delete them later.
+ // PinotFS has no batch move, so this is still one move per segment;
what this avoids relative to a
+ // per-segment helper is an exists() round trip per segment (move
already reports a missing source) and
+ // one log line per segment, both of which dominate when dropping a
table with many segments.
+ if (!filesToMove.isEmpty()) {
+ LOGGER.info("Moving {} segment files of table {} to the deleted
directory", filesToMove.size(),
+ tableNameWithType);
+ int movedCount = 0;
+ List<URI> failedMoves = new ArrayList<>();
+ for (int i = 0; i < filesToMove.size(); i++) {
+ URI srcUri = filesToMove.get(i);
+ URI destUri = moveDestinations.get(i);
+ try {
+ // Overwrites the file if it already exists in the target
directory.
+ if (pinotFS.move(srcUri, destUri, true)) {
+ // Touch is needed so that removeAgedDeletedSegments() sees a
fresh last-modified time.
+ // Only touch destinations we actually created: touching a path
whose move failed can
+ // materialize an empty object on object stores and leave a
phantom deleted segment behind.
+ if (!pinotFS.touch(destUri)) {
+ LOGGER.warn("Could not touch moved segment file at {}, it may
be aged out on the wrong schedule",
+ destUri);
+ }
+ movedCount++;
+ } else {
+ failedMoves.add(srcUri);
+ }
+ } catch (IOException e) {
+ failedMoves.add(srcUri);
+ LOGGER.debug("Could not move segment from {} to {}", srcUri,
destUri, e);
+ }
Review Comment:
`touch()` can throw `IOException` after `move()` has already succeeded (the
filesystem API and S3/GCS/HDFS implementations all allow this). The shared
catch then records the source as a failed move and omits it from `movedCount`,
producing an incorrect aggregate warning even though the source has moved.
Catch touch failures separately and still count the move as successful.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2552,84 +2571,173 @@ public void deleteTable(String tableName, TableType
tableType, @Nullable String
}
}
- // Remove the table from brokerResource
- HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+ // Acquire the deletion marker. It gives concurrent deletes mutual
exclusion and makes addTable() refuse to
+ // re-create the table while this deletion is still tearing its metadata
down.
+ if
(!ZKMetadataProvider.createOrTakeoverTableDeletionMarker(_propertyStore,
tableNameWithType, _controllerId)) {
+ // CONFLICT rather than a generic failure: a concurrent delete is an
expected, retryable condition.
+ throw new ControllerApplicationException(LOGGER, String.format(
+ "Cannot delete table '%s': a deletion is already in progress. "
+ + "If the previous deletion failed, wait for the deletion marker
to expire (24 hours) "
+ + "or manually clean up the deletion marker from ZK at path:
%s/%s",
+ tableNameWithType,
ZKMetadataProvider.getPropertyStoreTableDeletionInProgressPrefix(),
tableNameWithType),
+ Response.Status.CONFLICT);
+ }
+ try {
+ // Remove the table from brokerResource
+ HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+
+ // Delete the table on servers
+ deleteTableOnServers(tableNameWithType);
+
+ // Remove ideal state
+
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
+ LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+
+ // Remove all stored segments for the table.
+ // Retention precedence: the request's retention parameter, then the
table config, then the cluster default
+ // that SegmentDeletionManager applies when this is null. Reuse
SegmentDeletionManager's own extraction
+ // helper rather than re-reading the config here: it tolerates an empty
or malformed period, which
+ // TimeUtils.convertPeriodToMillis would otherwise throw on and make the
table undeletable.
+ Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod)
+ :
SegmentDeletionManager.getRetentionMsFromTableConfig(getTableConfig(tableNameWithType));
+ _segmentDeletionManager.removeSegmentsFromStoreInBatch(tableNameWithType,
+ getSegmentsFromPropertyStore(tableNameWithType),
+ retentionPeriodMs);
+ LOGGER.info("Deleting table {}: Removed stored segments",
tableNameWithType);
+
+ // Remove segment metadata
+
ZKMetadataProvider.removeResourceSegmentsFromPropertyStore(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed segment metadata",
tableNameWithType);
+
+ // Remove COMMITTING segment list
+ if (tableType == TableType.REALTIME) {
+ if (ZKMetadataProvider.removePauselessDebugMetadata(_propertyStore,
tableNameWithType)) {
+ LOGGER.info("Deleting table {}: Removed pauseless debug metadata",
tableNameWithType);
+ } else {
+ LOGGER.info("Deleting table {}: Failed to remove pauseless debug
metadata.", tableNameWithType);
+ }
+ }
- // Delete the table on servers
- deleteTableOnServers(tableNameWithType);
+ // Remove instance partitions
+ if (tableType == TableType.OFFLINE) {
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
tableNameWithType);
+ } else {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
+
InstancePartitionsType.CONSUMING.getInstancePartitionsName(rawTableName));
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
+
InstancePartitionsType.COMPLETED.getInstancePartitionsName(rawTableName));
+ }
+ LOGGER.info("Deleting table {}: Removed instance partitions",
tableNameWithType);
- // Remove ideal state
-
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
- LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+ // Remove tier instance partitions
+ InstancePartitionsUtils.removeTierInstancePartitions(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed tier instance partitions",
tableNameWithType);
- // Remove all stored segments for the table
- Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod) : null;
- _segmentDeletionManager.removeSegmentsFromStoreInBatch(tableNameWithType,
- getSegmentsFromPropertyStore(tableNameWithType),
- retentionPeriodMs);
- LOGGER.info("Deleting table {}: Removed stored segments",
tableNameWithType);
+ // Remove segment lineage
+ SegmentLineageAccessHelper.deleteSegmentLineage(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed segment lineage",
tableNameWithType);
- // Remove segment metadata
- ZKMetadataProvider.removeResourceSegmentsFromPropertyStore(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed segment metadata",
tableNameWithType);
+ // Remove task related metadata
+ MinionTaskMetadataUtils.deleteTaskMetadata(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed all minion task metadata",
tableNameWithType);
- // Remove COMMITTING segment list
- if (tableType == TableType.REALTIME) {
- if (ZKMetadataProvider.removePauselessDebugMetadata(_propertyStore,
tableNameWithType)) {
- LOGGER.info("Deleting table {}: Removed pauseless debug metadata",
tableNameWithType);
- } else {
- LOGGER.info("Deleting table {}: Failed to remove pauseless debug
metadata.", tableNameWithType);
+ // Remove materialized view metadata (if any) and unregister from
consistency manager
+ notifyMaterializedViewConsistencyManagerForTableDrop(tableNameWithType);
+ try {
+ MaterializedViewDefinitionMetadataUtils.delete(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed MV definition metadata",
tableNameWithType);
+ } catch (Exception e) {
+ LOGGER.debug("Deleting table {}: No MV definition metadata to remove
or removal failed",
+ tableNameWithType, e);
+ }
+ try {
+ MaterializedViewRuntimeMetadataUtils.delete(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed MV runtime metadata",
tableNameWithType);
+ } catch (Exception e) {
+ LOGGER.debug("Deleting table {}: No MV runtime metadata to remove or
removal failed",
+ tableNameWithType, e);
}
- }
- // Remove instance partitions
- if (tableType == TableType.OFFLINE) {
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
tableNameWithType);
- } else {
- String rawTableName = TableNameBuilder.extractRawTableName(tableName);
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
-
InstancePartitionsType.CONSUMING.getInstancePartitionsName(rawTableName));
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
-
InstancePartitionsType.COMPLETED.getInstancePartitionsName(rawTableName));
- }
- LOGGER.info("Deleting table {}: Removed instance partitions",
tableNameWithType);
+ // Remove table config
+ // NOTE: This should always be the last step for deletion to avoid race
condition in table re-create
+ ZKMetadataProvider.removeResourceConfigFromPropertyStore(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed table config",
tableNameWithType);
- // Remove tier instance partitions
- InstancePartitionsUtils.removeTierInstancePartitions(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed tier instance partitions",
tableNameWithType);
+ // Audit the property store for metadata that should now be gone, and
warn about anything left behind
+ List<String> leftoverMetadata =
findLeftoverTableMetadata(tableNameWithType, tableType);
+ if (leftoverMetadata.isEmpty()) {
+ LOGGER.info("Deleting table {}: Verified all audited metadata was
removed", tableNameWithType);
+ } else {
+ LOGGER.warn("Deleting table {}: Finished but left {} metadata
artifact(s) behind: {}. "
+ + "These need to be cleaned up manually before the table is
re-created.", tableNameWithType,
+ leftoverMetadata.size(), leftoverMetadata);
+ }
- // Remove segment lineage
- SegmentLineageAccessHelper.deleteSegmentLineage(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed segment lineage",
tableNameWithType);
+ LOGGER.info("Deleting table {}: Finish", tableNameWithType);
+ } finally {
+ // Always release the marker, even if the deletion failed, so the
deletion can be retried or the table
+ // re-created. Releasing is a no-op if another controller has since
taken the marker over.
+ ZKMetadataProvider.removeTableDeletionMarker(_propertyStore,
tableNameWithType, _controllerId);
+ }
+ }
- // Remove task related metadata
- MinionTaskMetadataUtils.deleteTaskMetadata(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed all minion task metadata",
tableNameWithType);
+ /// Returns the table metadata that [#deleteTable] should have removed but
which is still present, so the
+ /// caller can warn an operator about it. An empty list means the audited
metadata is all gone.
+ ///
+ /// This deliberately only covers artifacts that `deleteTable` removes with
a synchronous ZK write, so a
+ /// leftover here is a real leak rather than a timing artifact. In
particular the ExternalView is NOT checked:
+ /// Helix removes it asynchronously via the controller pipeline after the
IdealState is dropped, so it is
+ /// routinely still present at this point.
+ ///
+ /// The caller only warns and never throws. By the time this runs the table
config is already gone, so the
+ /// delete has succeeded from the caller's point of view; failing the
request would report a misleading error
+ /// for a table that no longer exists, and would not make the leftovers go
away.
+ ///
+ /// ponytail: tier instance partitions, minion task metadata and
materialized view metadata are not audited.
Review Comment:
“ponytail” is unrelated to the intended caveat and makes this Javadoc
unclear; use “Note” here.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2552,84 +2571,173 @@ public void deleteTable(String tableName, TableType
tableType, @Nullable String
}
}
- // Remove the table from brokerResource
- HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+ // Acquire the deletion marker. It gives concurrent deletes mutual
exclusion and makes addTable() refuse to
+ // re-create the table while this deletion is still tearing its metadata
down.
+ if
(!ZKMetadataProvider.createOrTakeoverTableDeletionMarker(_propertyStore,
tableNameWithType, _controllerId)) {
+ // CONFLICT rather than a generic failure: a concurrent delete is an
expected, retryable condition.
+ throw new ControllerApplicationException(LOGGER, String.format(
+ "Cannot delete table '%s': a deletion is already in progress. "
+ + "If the previous deletion failed, wait for the deletion marker
to expire (24 hours) "
+ + "or manually clean up the deletion marker from ZK at path:
%s/%s",
+ tableNameWithType,
ZKMetadataProvider.getPropertyStoreTableDeletionInProgressPrefix(),
tableNameWithType),
+ Response.Status.CONFLICT);
+ }
+ try {
+ // Remove the table from brokerResource
+ HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+
+ // Delete the table on servers
+ deleteTableOnServers(tableNameWithType);
+
+ // Remove ideal state
+
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
+ LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+
+ // Remove all stored segments for the table.
+ // Retention precedence: the request's retention parameter, then the
table config, then the cluster default
+ // that SegmentDeletionManager applies when this is null. Reuse
SegmentDeletionManager's own extraction
+ // helper rather than re-reading the config here: it tolerates an empty
or malformed period, which
+ // TimeUtils.convertPeriodToMillis would otherwise throw on and make the
table undeletable.
+ Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod)
+ :
SegmentDeletionManager.getRetentionMsFromTableConfig(getTableConfig(tableNameWithType));
+ _segmentDeletionManager.removeSegmentsFromStoreInBatch(tableNameWithType,
+ getSegmentsFromPropertyStore(tableNameWithType),
+ retentionPeriodMs);
+ LOGGER.info("Deleting table {}: Removed stored segments",
tableNameWithType);
+
+ // Remove segment metadata
+
ZKMetadataProvider.removeResourceSegmentsFromPropertyStore(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed segment metadata",
tableNameWithType);
+
+ // Remove COMMITTING segment list
+ if (tableType == TableType.REALTIME) {
+ if (ZKMetadataProvider.removePauselessDebugMetadata(_propertyStore,
tableNameWithType)) {
+ LOGGER.info("Deleting table {}: Removed pauseless debug metadata",
tableNameWithType);
+ } else {
+ LOGGER.info("Deleting table {}: Failed to remove pauseless debug
metadata.", tableNameWithType);
+ }
+ }
- // Delete the table on servers
- deleteTableOnServers(tableNameWithType);
+ // Remove instance partitions
+ if (tableType == TableType.OFFLINE) {
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
tableNameWithType);
+ } else {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
+
InstancePartitionsType.CONSUMING.getInstancePartitionsName(rawTableName));
+ InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
+
InstancePartitionsType.COMPLETED.getInstancePartitionsName(rawTableName));
+ }
+ LOGGER.info("Deleting table {}: Removed instance partitions",
tableNameWithType);
- // Remove ideal state
-
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
- LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+ // Remove tier instance partitions
+ InstancePartitionsUtils.removeTierInstancePartitions(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed tier instance partitions",
tableNameWithType);
- // Remove all stored segments for the table
- Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod) : null;
- _segmentDeletionManager.removeSegmentsFromStoreInBatch(tableNameWithType,
- getSegmentsFromPropertyStore(tableNameWithType),
- retentionPeriodMs);
- LOGGER.info("Deleting table {}: Removed stored segments",
tableNameWithType);
+ // Remove segment lineage
+ SegmentLineageAccessHelper.deleteSegmentLineage(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed segment lineage",
tableNameWithType);
- // Remove segment metadata
- ZKMetadataProvider.removeResourceSegmentsFromPropertyStore(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed segment metadata",
tableNameWithType);
+ // Remove task related metadata
+ MinionTaskMetadataUtils.deleteTaskMetadata(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed all minion task metadata",
tableNameWithType);
- // Remove COMMITTING segment list
- if (tableType == TableType.REALTIME) {
- if (ZKMetadataProvider.removePauselessDebugMetadata(_propertyStore,
tableNameWithType)) {
- LOGGER.info("Deleting table {}: Removed pauseless debug metadata",
tableNameWithType);
- } else {
- LOGGER.info("Deleting table {}: Failed to remove pauseless debug
metadata.", tableNameWithType);
+ // Remove materialized view metadata (if any) and unregister from
consistency manager
+ notifyMaterializedViewConsistencyManagerForTableDrop(tableNameWithType);
+ try {
+ MaterializedViewDefinitionMetadataUtils.delete(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed MV definition metadata",
tableNameWithType);
+ } catch (Exception e) {
+ LOGGER.debug("Deleting table {}: No MV definition metadata to remove
or removal failed",
+ tableNameWithType, e);
+ }
+ try {
+ MaterializedViewRuntimeMetadataUtils.delete(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed MV runtime metadata",
tableNameWithType);
+ } catch (Exception e) {
+ LOGGER.debug("Deleting table {}: No MV runtime metadata to remove or
removal failed",
+ tableNameWithType, e);
}
- }
- // Remove instance partitions
- if (tableType == TableType.OFFLINE) {
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
tableNameWithType);
- } else {
- String rawTableName = TableNameBuilder.extractRawTableName(tableName);
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
-
InstancePartitionsType.CONSUMING.getInstancePartitionsName(rawTableName));
- InstancePartitionsUtils.removeInstancePartitions(_propertyStore,
-
InstancePartitionsType.COMPLETED.getInstancePartitionsName(rawTableName));
- }
- LOGGER.info("Deleting table {}: Removed instance partitions",
tableNameWithType);
+ // Remove table config
+ // NOTE: This should always be the last step for deletion to avoid race
condition in table re-create
+ ZKMetadataProvider.removeResourceConfigFromPropertyStore(_propertyStore,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed table config",
tableNameWithType);
- // Remove tier instance partitions
- InstancePartitionsUtils.removeTierInstancePartitions(_propertyStore,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed tier instance partitions",
tableNameWithType);
+ // Audit the property store for metadata that should now be gone, and
warn about anything left behind
+ List<String> leftoverMetadata =
findLeftoverTableMetadata(tableNameWithType, tableType);
+ if (leftoverMetadata.isEmpty()) {
+ LOGGER.info("Deleting table {}: Verified all audited metadata was
removed", tableNameWithType);
+ } else {
+ LOGGER.warn("Deleting table {}: Finished but left {} metadata
artifact(s) behind: {}. "
+ + "These need to be cleaned up manually before the table is
re-created.", tableNameWithType,
+ leftoverMetadata.size(), leftoverMetadata);
+ }
Review Comment:
The audit is not actually warn-only: any transient Helix/property-store
exception from `findLeftoverTableMetadata` escapes after the table config has
been removed, so the endpoint returns a misleading 500 despite the stated
contract. Catch audit failures here and log them without changing the completed
deletion result.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2552,84 +2571,173 @@ public void deleteTable(String tableName, TableType
tableType, @Nullable String
}
}
- // Remove the table from brokerResource
- HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+ // Acquire the deletion marker. It gives concurrent deletes mutual
exclusion and makes addTable() refuse to
+ // re-create the table while this deletion is still tearing its metadata
down.
+ if
(!ZKMetadataProvider.createOrTakeoverTableDeletionMarker(_propertyStore,
tableNameWithType, _controllerId)) {
+ // CONFLICT rather than a generic failure: a concurrent delete is an
expected, retryable condition.
+ throw new ControllerApplicationException(LOGGER, String.format(
+ "Cannot delete table '%s': a deletion is already in progress. "
+ + "If the previous deletion failed, wait for the deletion marker
to expire (24 hours) "
+ + "or manually clean up the deletion marker from ZK at path:
%s/%s",
+ tableNameWithType,
ZKMetadataProvider.getPropertyStoreTableDeletionInProgressPrefix(),
tableNameWithType),
+ Response.Status.CONFLICT);
+ }
+ try {
+ // Remove the table from brokerResource
+ HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+
+ // Delete the table on servers
+ deleteTableOnServers(tableNameWithType);
+
+ // Remove ideal state
+
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
+ LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+
+ // Remove all stored segments for the table.
+ // Retention precedence: the request's retention parameter, then the
table config, then the cluster default
+ // that SegmentDeletionManager applies when this is null. Reuse
SegmentDeletionManager's own extraction
+ // helper rather than re-reading the config here: it tolerates an empty
or malformed period, which
+ // TimeUtils.convertPeriodToMillis would otherwise throw on and make the
table undeletable.
+ Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod)
+ :
SegmentDeletionManager.getRetentionMsFromTableConfig(getTableConfig(tableNameWithType));
Review Comment:
This changes the default retention precedence, but the public DELETE-table
`@ApiParam` and `DeleteTableCommand` help still say an omitted value goes
directly to the cluster setting and then `7d`. Update both user-facing
descriptions to include the table config first, as the segment-deletion API
already does.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2552,84 +2571,173 @@ public void deleteTable(String tableName, TableType
tableType, @Nullable String
}
}
- // Remove the table from brokerResource
- HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
- LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+ // Acquire the deletion marker. It gives concurrent deletes mutual
exclusion and makes addTable() refuse to
+ // re-create the table while this deletion is still tearing its metadata
down.
+ if
(!ZKMetadataProvider.createOrTakeoverTableDeletionMarker(_propertyStore,
tableNameWithType, _controllerId)) {
+ // CONFLICT rather than a generic failure: a concurrent delete is an
expected, retryable condition.
+ throw new ControllerApplicationException(LOGGER, String.format(
+ "Cannot delete table '%s': a deletion is already in progress. "
+ + "If the previous deletion failed, wait for the deletion marker
to expire (24 hours) "
+ + "or manually clean up the deletion marker from ZK at path:
%s/%s",
+ tableNameWithType,
ZKMetadataProvider.getPropertyStoreTableDeletionInProgressPrefix(),
tableNameWithType),
+ Response.Status.CONFLICT);
+ }
+ try {
+ // Remove the table from brokerResource
+ HelixHelper.removeResourceFromBrokerIdealState(_helixZkManager,
tableNameWithType);
+ LOGGER.info("Deleting table {}: Removed from broker resource",
tableNameWithType);
+
+ // Delete the table on servers
+ deleteTableOnServers(tableNameWithType);
+
+ // Remove ideal state
+
_helixDataAccessor.removeProperty(_keyBuilder.idealStates(tableNameWithType));
+ LOGGER.info("Deleting table {}: Removed ideal state", tableNameWithType);
+
+ // Remove all stored segments for the table.
+ // Retention precedence: the request's retention parameter, then the
table config, then the cluster default
+ // that SegmentDeletionManager applies when this is null. Reuse
SegmentDeletionManager's own extraction
+ // helper rather than re-reading the config here: it tolerates an empty
or malformed period, which
+ // TimeUtils.convertPeriodToMillis would otherwise throw on and make the
table undeletable.
+ Long retentionPeriodMs = retentionPeriod != null ?
TimeUtils.convertPeriodToMillis(retentionPeriod)
+ :
SegmentDeletionManager.getRetentionMsFromTableConfig(getTableConfig(tableNameWithType));
Review Comment:
The new table-config retention fallback has no regression test in the
controller tests. Add a deletion test with no request retention and a
distinctive `deletedSegmentsRetentionPeriod`, then verify that value reaches
the segment move naming/manager; also cover that an explicit request value
still takes precedence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
