This is an automated email from the ASF dual-hosted git repository.
manishswaminathan 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 7e4bc52158a UCT: Add getValidDocIdsType method for UpsertConfig
handling and validation (#16812)
7e4bc52158a is described below
commit 7e4bc52158ad0b9695c4c70418280fbf5e89bd05
Author: Abhishek Bafna <[email protected]>
AuthorDate: Wed Sep 17 14:01:56 2025 +0530
UCT: Add getValidDocIdsType method for UpsertConfig handling and validation
(#16812)
* UCT: Add getValidDocIdsType method for UpsertConfig handling and
validation
* addressing review comments
* Fix lint issue
* update integration test
---------
Co-authored-by: abhishekbafna <[email protected]>
---
.../tests/UpsertTableIntegrationTest.java | 2 +-
.../pinot/plugin/minion/tasks/MinionTaskUtils.java | 48 +++++++
.../UpsertCompactionTaskExecutor.java | 5 +-
.../UpsertCompactionTaskGenerator.java | 35 ++---
.../plugin/minion/tasks/MinionTaskUtilsTest.java | 158 ++++++++++++++++++++-
.../UpsertCompactionTaskGeneratorTest.java | 83 +++++++++++
6 files changed, 303 insertions(+), 28 deletions(-)
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java
index 40f57e2583c..8c01f6601d5 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java
@@ -491,7 +491,7 @@ public class UpsertTableIntegrationTest extends
BaseClusterIntegrationTest {
TableConfig tableConfig = setUpTable(tableName, kafkaTopicName,
upsertConfig);
TableTaskConfig taskConfig = getCompactionTaskConfig();
taskConfig.getConfigsForTaskType(MinionConstants.UpsertCompactionTask.TASK_TYPE)
- .put("validDocIdsType", ValidDocIdsType.IN_MEMORY.name());
+ .put("validDocIdsType", ValidDocIdsType.IN_MEMORY_WITH_DELETE.name());
tableConfig.setTaskConfig(taskConfig);
updateTableConfig(tableConfig);
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java
index a8be2803544..91914dc802b 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.plugin.minion.tasks;
+import com.google.common.base.Preconditions;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.time.Instant;
@@ -28,10 +29,12 @@ import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.helix.HelixAdmin;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.restlet.resources.ValidDocIdsBitmapResponse;
+import org.apache.pinot.common.restlet.resources.ValidDocIdsType;
import org.apache.pinot.common.utils.RoaringBitmapUtils;
import org.apache.pinot.common.utils.ServiceStatus;
import org.apache.pinot.common.utils.config.InstanceUtils;
@@ -40,6 +43,7 @@ import
org.apache.pinot.controller.util.ServerSegmentMetadataReader;
import org.apache.pinot.minion.MinionContext;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
+import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.filesystem.LocalPinotFS;
import org.apache.pinot.spi.filesystem.PinotFS;
@@ -47,6 +51,7 @@ import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.roaringbitmap.RoaringBitmap;
@@ -265,4 +270,47 @@ public class MinionTaskUtils {
public static long fromUTCString(String utcString) {
return Instant.parse(utcString).toEpochMilli();
}
+
+ /**
+ * Get the validDocIdsType based on the upsertConfig and taskConfigs.
+ * The default value is determined by whether delete is enabled in the
upsertConfig. If delete is enabled,
+ * the default value is 'snapshot_with_delete', otherwise it is 'snapshot'.
+ * If delete is enabled, we override the user-specified value to
'snapshot_with_delete' for backward compatibility
+ * except when it is 'in_memory_with_delete'.
+ * It also validates the combination of validDocIdsType, snapshot and
deleteRecordColumn.
+ * @param upsertConfig upsertConfig of the table
+ * @param taskConfigs taskConfigs of the task
+ * @param validDocIdsTypeKey the key to get validDocIdsType from taskConfigs
+ * @return the validDocIdsType
+ */
+ public static ValidDocIdsType getValidDocIdsType(UpsertConfig upsertConfig,
Map<String, String> taskConfigs,
+ String validDocIdsTypeKey) {
+ boolean isDeleteEnabled =
StringUtils.isNotEmpty(upsertConfig.getDeleteRecordColumn());
+ ValidDocIdsType defaultValidDocIdsType =
+ isDeleteEnabled ? ValidDocIdsType.SNAPSHOT_WITH_DELETE :
ValidDocIdsType.SNAPSHOT;
+ String validDocIdsTypeStr = taskConfigs.getOrDefault(validDocIdsTypeKey,
+ defaultValidDocIdsType.name()).toUpperCase();
+ ValidDocIdsType validDocIdsType =
ValidDocIdsType.valueOf(validDocIdsTypeStr);
+
+ if (isDeleteEnabled && validDocIdsType !=
ValidDocIdsType.SNAPSHOT_WITH_DELETE
+ && validDocIdsType != ValidDocIdsType.IN_MEMORY_WITH_DELETE) {
+ LOGGER.warn(
+ "Overriding user-specified validDocIdsType '{}' to '{}' for backward
compatibility because delete is "
+ + "enabled (deleteRecordColumn='{}').",
+ validDocIdsType, ValidDocIdsType.SNAPSHOT_WITH_DELETE,
upsertConfig.getDeleteRecordColumn());
+ validDocIdsType = ValidDocIdsType.SNAPSHOT_WITH_DELETE;
+ }
+
+ if (validDocIdsType == ValidDocIdsType.SNAPSHOT || validDocIdsType ==
ValidDocIdsType.SNAPSHOT_WITH_DELETE) {
+ Preconditions.checkState(upsertConfig.getSnapshot() !=
Enablement.DISABLE,
+ "'snapshot' must not be 'DISABLE' with validDocIdsType: %s",
validDocIdsType);
+ }
+
+ if (validDocIdsType == ValidDocIdsType.IN_MEMORY_WITH_DELETE
+ || validDocIdsType == ValidDocIdsType.SNAPSHOT_WITH_DELETE) {
+ Preconditions.checkState(isDeleteEnabled,
+ "'deleteRecordColumn' must be provided with validDocIdsType: %s",
validDocIdsType);
+ }
+ return validDocIdsType;
+ }
}
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskExecutor.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskExecutor.java
index 1766d1c6646..490387f727d 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskExecutor.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskExecutor.java
@@ -24,7 +24,6 @@ import java.util.Map;
import org.apache.commons.io.FileUtils;
import
org.apache.pinot.common.metadata.segment.SegmentZKMetadataCustomMapModifier;
import org.apache.pinot.common.metrics.MinionMeter;
-import org.apache.pinot.common.restlet.resources.ValidDocIdsType;
import org.apache.pinot.core.common.MinionConstants;
import org.apache.pinot.core.common.MinionConstants.UpsertCompactionTask;
import org.apache.pinot.core.minion.PinotTaskConfig;
@@ -61,8 +60,8 @@ public class UpsertCompactionTaskExecutor extends
BaseSingleSegmentConversionExe
String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY);
TableConfig tableConfig = getTableConfig(tableNameWithType);
- String validDocIdsTypeStr =
- configs.getOrDefault(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
ValidDocIdsType.SNAPSHOT.name());
+ String validDocIdsTypeStr =
MinionTaskUtils.getValidDocIdsType(tableConfig.getUpsertConfig(), configs,
+ UpsertCompactionTask.VALID_DOC_IDS_TYPE).toString();
SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
String originalSegmentCrcFromTaskGenerator =
configs.get(MinionConstants.ORIGINAL_SEGMENT_CRC_KEY);
String crcFromDeepStorageSegment = segmentMetadata.getCrc();
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGenerator.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGenerator.java
index 79baaef7bf4..27632ca182d 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGenerator.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGenerator.java
@@ -42,13 +42,13 @@ import
org.apache.pinot.controller.util.ServerSegmentMetadataReader;
import org.apache.pinot.core.common.MinionConstants;
import org.apache.pinot.core.common.MinionConstants.UpsertCompactionTask;
import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.plugin.minion.tasks.MinionTaskUtils;
import org.apache.pinot.spi.annotations.minion.TaskGenerator;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -102,6 +102,11 @@ public class UpsertCompactionTaskGenerator extends
BaseTaskGenerator {
continue;
}
+ if (!tableConfig.isUpsertEnabled()) {
+ LOGGER.warn("Upsert config is not enabled for table: {}",
tableNameWithType);
+ continue;
+ }
+
Map<String, String> taskConfigs =
tableConfig.getTaskConfig().getConfigsForTaskType(taskType);
List<SegmentZKMetadata> allSegments =
_clusterInfoAccessor.getSegmentsZKMetadata(tableNameWithType);
@@ -138,11 +143,8 @@ public class UpsertCompactionTaskGenerator extends
BaseTaskGenerator {
new ServerSegmentMetadataReader(_clusterInfoAccessor.getExecutor(),
_clusterInfoAccessor.getConnectionManager());
- // By default, we use 'snapshot' for validDocIdsType. This means that we
will use the validDocIds bitmap from
- // the snapshot from Pinot segment. This will require 'enableSnapshot'
from UpsertConfig to be set to true.
- String validDocIdsTypeStr =
- taskConfigs.getOrDefault(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
ValidDocIdsType.SNAPSHOT.toString());
- ValidDocIdsType validDocIdsType =
ValidDocIdsType.valueOf(validDocIdsTypeStr.toUpperCase());
+ ValidDocIdsType validDocIdsType =
MinionTaskUtils.getValidDocIdsType(tableConfig.getUpsertConfig(), taskConfigs,
+ UpsertCompactionTask.VALID_DOC_IDS_TYPE);
// Number of segments to query per server request. If a table has a lot
of segments, then we might send a
// huge payload to pinot-server in request. Batching the requests will
help in reducing the payload size.
@@ -317,21 +319,10 @@ public class UpsertCompactionTaskGenerator extends
BaseTaskGenerator {
taskConfigs.containsKey(UpsertCompactionTask.INVALID_RECORDS_THRESHOLD_PERCENT)
|| taskConfigs.containsKey(
UpsertCompactionTask.INVALID_RECORDS_THRESHOLD_COUNT),
"invalidRecordsThresholdPercent or invalidRecordsThresholdCount or
both must be provided");
- String validDocIdsType =
- taskConfigs.getOrDefault(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
UpsertCompactionTask.SNAPSHOT);
- if (validDocIdsType.equals(ValidDocIdsType.SNAPSHOT.toString())) {
- UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- assert upsertConfig != null;
- // NOTE: Allow snapshot to be DEFAULT because it might be enabled at
server level.
- Preconditions.checkState(upsertConfig.getSnapshot() !=
Enablement.DISABLE,
- "'snapshot' from UpsertConfig must not be 'DISABLE' for
UpsertCompactionTask with validDocIdsType = %s",
- validDocIdsType);
- } else if
(validDocIdsType.equals(ValidDocIdsType.IN_MEMORY_WITH_DELETE.toString())) {
- UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
- assert upsertConfig != null;
- Preconditions.checkNotNull(upsertConfig.getDeleteRecordColumn(),
String.format(
- "deleteRecordColumn must be provided for " + "UpsertCompactionTask
with validDocIdsType = %s",
- validDocIdsType));
- }
+
+ // validate validDocIdsType default logic
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
+ assert upsertConfig != null;
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
}
}
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtilsTest.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtilsTest.java
index c2456a9957f..a5ff73f8063 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtilsTest.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtilsTest.java
@@ -22,24 +22,30 @@ import java.net.URI;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.apache.pinot.common.restlet.resources.ValidDocIdsType;
import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.common.MinionConstants.UpsertCompactionTask;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.filesystem.LocalPinotFS;
import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class MinionTaskUtilsTest {
@Test
- public void testGetInputPinotFS() throws Exception {
+ public void testGetInputPinotFS()
+ throws Exception {
Map<String, String> taskConfigs = new HashMap<>();
taskConfigs.put("input.fs.className",
"org.apache.pinot.spi.filesystem.LocalPinotFS");
URI fileURI = new URI("file:///path/to/file");
@@ -50,7 +56,8 @@ public class MinionTaskUtilsTest {
}
@Test
- public void testGetOutputPinotFS() throws Exception {
+ public void testGetOutputPinotFS()
+ throws Exception {
Map<String, String> taskConfigs = new HashMap<>();
taskConfigs.put("output.fs.className",
"org.apache.pinot.spi.filesystem.LocalPinotFS");
URI fileURI = new URI("file:///path/to/file");
@@ -105,4 +112,151 @@ public class MinionTaskUtilsTest {
assertFalse(MinionTaskUtils.extractMinionAllowDownloadFromServer(tableConfig,
MinionConstants.MergeRollupTask.TASK_TYPE, false));
}
+
+ @Test
+ public void testGetValidDocIdsTypeDefaultBehavior() {
+ // Test default behavior scenarios
+ UpsertConfig upsertConfig = new UpsertConfig();
+ Map<String, String> taskConfigs = new HashMap<>();
+
+ // Test 1: Default when delete is not enabled
+ ValidDocIdsType result1 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result1, ValidDocIdsType.SNAPSHOT);
+
+ // Test 2: Default when delete is enabled
+ upsertConfig.setDeleteRecordColumn("deleted");
+ ValidDocIdsType result2 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result2, ValidDocIdsType.SNAPSHOT_WITH_DELETE);
+ }
+
+ @Test
+ public void testGetValidDocIdsTypeExplicitConfigurations() {
+ // Test explicit configuration scenarios
+ UpsertConfig upsertConfig = new UpsertConfig();
+ Map<String, String> taskConfigs = new HashMap<>();
+
+ // Test 1: Explicit SNAPSHOT with enabled snapshot
+ upsertConfig.setSnapshot(Enablement.ENABLE);
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "SNAPSHOT");
+ ValidDocIdsType result1 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result1, ValidDocIdsType.SNAPSHOT);
+
+ // Test 2: Explicit SNAPSHOT with default snapshot enablement
+ upsertConfig = new UpsertConfig(); // Reset to default
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "SNAPSHOT");
+ ValidDocIdsType result2 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result2, ValidDocIdsType.SNAPSHOT);
+
+ // Test 3: Explicit IN_MEMORY without delete
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "IN_MEMORY");
+ ValidDocIdsType result3 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result3, ValidDocIdsType.IN_MEMORY);
+
+ // Test 4: Explicit IN_MEMORY_WITH_DELETE with delete column
+ upsertConfig.setDeleteRecordColumn("deleted");
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"IN_MEMORY_WITH_DELETE");
+ ValidDocIdsType result4 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result4, ValidDocIdsType.IN_MEMORY_WITH_DELETE);
+
+ // Test 5: Explicit SNAPSHOT_WITH_DELETE with delete column
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"SNAPSHOT_WITH_DELETE");
+ ValidDocIdsType result5 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result5, ValidDocIdsType.SNAPSHOT_WITH_DELETE);
+ }
+
+ @Test
+ public void testGetValidDocIdsTypeBackwardCompatibilityAndOverride() {
+ // Test backward compatibility behavior when delete is enabled
+ UpsertConfig upsertConfig = new UpsertConfig();
+ upsertConfig.setDeleteRecordColumn("deleted");
+ Map<String, String> taskConfigs = new HashMap<>();
+
+ // Test 1: SNAPSHOT gets overridden to SNAPSHOT_WITH_DELETE
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "SNAPSHOT");
+ ValidDocIdsType result1 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result1, ValidDocIdsType.SNAPSHOT_WITH_DELETE);
+
+ // Test 2: IN_MEMORY gets overridden to SNAPSHOT_WITH_DELETE
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "IN_MEMORY");
+ ValidDocIdsType result2 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result2, ValidDocIdsType.SNAPSHOT_WITH_DELETE);
+
+ // Test 3: SNAPSHOT_WITH_DELETE stays the same
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"SNAPSHOT_WITH_DELETE");
+ ValidDocIdsType result3 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result3, ValidDocIdsType.SNAPSHOT_WITH_DELETE);
+
+ // Test 4: IN_MEMORY_WITH_DELETE stays the same (not overridden)
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"IN_MEMORY_WITH_DELETE");
+ ValidDocIdsType result4 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result4, ValidDocIdsType.IN_MEMORY_WITH_DELETE);
+
+ // Test 5: Case insensitive override behavior
+ taskConfigs.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"in_memory_with_delete");
+ ValidDocIdsType result5 =
+ MinionTaskUtils.getValidDocIdsType(upsertConfig, taskConfigs,
UpsertCompactionTask.VALID_DOC_IDS_TYPE);
+ assertEquals(result5, ValidDocIdsType.IN_MEMORY_WITH_DELETE);
+ }
+
+ @Test
+ public void testGetValidDocIdsTypeValidationErrors() {
+ // Test validation error scenarios
+
+ // Test 1: SNAPSHOT with disabled snapshot
+ UpsertConfig upsertConfig1 = new UpsertConfig();
+ upsertConfig1.setSnapshot(Enablement.DISABLE);
+ Map<String, String> taskConfigs1 = new HashMap<>();
+ taskConfigs1.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE, "SNAPSHOT");
+ IllegalStateException exception1 = expectThrows(
+ "Expected IllegalStateException when using SNAPSHOT with snapshot
disabled",
+ IllegalStateException.class,
+ () -> MinionTaskUtils.getValidDocIdsType(upsertConfig1, taskConfigs1,
UpsertCompactionTask.VALID_DOC_IDS_TYPE));
+ assertEquals(exception1.getMessage(), "'snapshot' must not be 'DISABLE'
with validDocIdsType: SNAPSHOT");
+
+ // Test 2: IN_MEMORY_WITH_DELETE without delete column
+ UpsertConfig upsertConfig2 = new UpsertConfig();
+ Map<String, String> taskConfigs2 = new HashMap<>();
+ taskConfigs2.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"IN_MEMORY_WITH_DELETE");
+ IllegalStateException exception2 = expectThrows(
+ "Expected IllegalStateException when using IN_MEMORY_WITH_DELETE
without delete column",
+ IllegalStateException.class,
+ () -> MinionTaskUtils.getValidDocIdsType(upsertConfig2, taskConfigs2,
UpsertCompactionTask.VALID_DOC_IDS_TYPE));
+ assertEquals(exception2.getMessage(),
+ "'deleteRecordColumn' must be provided with validDocIdsType:
IN_MEMORY_WITH_DELETE");
+
+ // Test 3: SNAPSHOT_WITH_DELETE without delete column
+ UpsertConfig upsertConfig3 = new UpsertConfig();
+ Map<String, String> taskConfigs3 = new HashMap<>();
+ taskConfigs3.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"SNAPSHOT_WITH_DELETE");
+ IllegalStateException exception3 = expectThrows(
+ "Expected IllegalStateException when using SNAPSHOT_WITH_DELETE
without delete column",
+ IllegalStateException.class,
+ () -> MinionTaskUtils.getValidDocIdsType(upsertConfig3, taskConfigs3,
UpsertCompactionTask.VALID_DOC_IDS_TYPE));
+ assertEquals(exception3.getMessage(),
+ "'deleteRecordColumn' must be provided with validDocIdsType:
SNAPSHOT_WITH_DELETE");
+
+ // Test 4: SNAPSHOT_WITH_DELETE with disabled snapshot
+ UpsertConfig upsertConfig4 = new UpsertConfig();
+ upsertConfig4.setDeleteRecordColumn("deleted");
+ upsertConfig4.setSnapshot(Enablement.DISABLE);
+ Map<String, String> taskConfigs4 = new HashMap<>();
+ taskConfigs4.put(UpsertCompactionTask.VALID_DOC_IDS_TYPE,
"SNAPSHOT_WITH_DELETE");
+ IllegalStateException exception4 = expectThrows(
+ "Expected IllegalStateException when using SNAPSHOT_WITH_DELETE with
snapshot disabled",
+ IllegalStateException.class,
+ () -> MinionTaskUtils.getValidDocIdsType(upsertConfig4, taskConfigs4,
UpsertCompactionTask.VALID_DOC_IDS_TYPE));
+ assertEquals(exception4.getMessage(),
+ "'snapshot' must not be 'DISABLE' with validDocIdsType:
SNAPSHOT_WITH_DELETE");
+ }
}
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGeneratorTest.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGeneratorTest.java
index e94dae11fc1..8879a8e4399 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGeneratorTest.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/upsertcompaction/UpsertCompactionTaskGeneratorTest.java
@@ -387,6 +387,89 @@ public class UpsertCompactionTaskGeneratorTest {
() -> _taskGenerator.validateTaskConfigs(invalidTableConfig, new
Schema(), upsertCompactionTaskConfig5));
}
+ @Test
+ public void testGenerateNoTaskWhenUpsertIsNotEnabled() {
+ // Create a table config without upsert enabled
+ TableConfig tableConfigWithoutUpsert =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName(TIME_COLUMN_NAME)
+ .build();
+
+ // Mock cluster info accessor with completed segments
+
when(_mockClusterInfoAccessor.getSegmentsZKMetadata(REALTIME_TABLE_NAME)).thenReturn(
+ Lists.newArrayList(_completedSegment, _completedSegment2));
+
when(_mockClusterInfoAccessor.getIdealState(REALTIME_TABLE_NAME)).thenReturn(getIdealState(REALTIME_TABLE_NAME,
+ Lists.newArrayList(_completedSegment.getSegmentName(),
_completedSegment2.getSegmentName())));
+
+ _taskGenerator.init(_mockClusterInfoAccessor);
+
+ // Generate tasks for table without upsert - should return empty list
+ List<PinotTaskConfig> pinotTaskConfigs =
_taskGenerator.generateTasks(Lists.newArrayList(tableConfigWithoutUpsert));
+
+ // Verify no tasks are generated when upsert is not enabled
+ assertEquals(pinotTaskConfigs.size(), 0);
+ }
+
+ @Test
+ public void testValidateTaskConfigsValidDocIdsTypeSuccess() {
+ // Test valid configurations for different ValidDocIdsType values
+ testValidDocIdsTypeConfiguration("SNAPSHOT_WITH_DELETE", true, true);
+ testValidDocIdsTypeConfiguration("IN_MEMORY_WITH_DELETE", false, true);
+ testValidDocIdsTypeConfiguration("SNAPSHOT", true, false);
+ testValidDocIdsTypeConfiguration("IN_MEMORY", false, false);
+ }
+
+ @Test
+ public void testValidateTaskConfigsValidDocIdsTypeErrors() {
+ // Test error scenarios for ValidDocIdsType validation
+ testValidDocIdsTypeError("SNAPSHOT_WITH_DELETE", false, true,
+ "'snapshot' must not be 'DISABLE'");
+ testValidDocIdsTypeError("SNAPSHOT_WITH_DELETE", true, false,
+ "'deleteRecordColumn' must be provided");
+ testValidDocIdsTypeError("IN_MEMORY_WITH_DELETE", false, false,
+ "'deleteRecordColumn' must be provided");
+ }
+
+ private void testValidDocIdsTypeConfiguration(String validDocIdsType,
boolean enableSnapshot,
+ boolean setDeleteColumn) {
+ TableConfig tableConfig = getTableConfig(validDocIdsType, enableSnapshot,
setDeleteColumn);
+ Map<String, String> taskConfigs =
tableConfig.getTaskConfig().getConfigsForTaskType("UpsertCompactionTask");
+
+ // Should not throw exception for valid configuration
+ _taskGenerator.validateTaskConfigs(tableConfig, new Schema(), taskConfigs);
+ }
+
+ private void testValidDocIdsTypeError(String validDocIdsType, boolean
enableSnapshot,
+ boolean setDeleteColumn, String expectedError) {
+ TableConfig tableConfig = getTableConfig(validDocIdsType, enableSnapshot,
setDeleteColumn);
+ Map<String, String> taskConfigs =
tableConfig.getTaskConfig().getConfigsForTaskType("UpsertCompactionTask");
+
+ // Should throw exception with expected error message
+ IllegalStateException exception =
Assert.expectThrows(IllegalStateException.class,
+ () -> _taskGenerator.validateTaskConfigs(tableConfig, new Schema(),
taskConfigs));
+ assertTrue(exception.getMessage().contains(expectedError),
exception.getMessage());
+ }
+
+ private static TableConfig getTableConfig(String validDocIdsType, boolean
enableSnapshot, boolean setDeleteColumn) {
+ UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
+
+ if (setDeleteColumn) {
+ upsertConfig.setDeleteRecordColumn("deleted");
+ }
+
+ upsertConfig.setSnapshot(enableSnapshot ? Enablement.ENABLE :
Enablement.DISABLE);
+
+ Map<String, String> taskConfigs = new HashMap<>();
+ taskConfigs.put("invalidRecordsThresholdPercent", "10");
+ taskConfigs.put("validDocIdsType", validDocIdsType);
+
+ TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME)
+ .setTableName(RAW_TABLE_NAME)
+ .setUpsertConfig(upsertConfig)
+ .setTaskConfig(new
TableTaskConfig(ImmutableMap.of("UpsertCompactionTask", taskConfigs)))
+ .build();
+ return tableConfig;
+ }
+
private Map<String, String> getCompactionConfigs(String
invalidRecordsThresholdPercent,
String invalidRecordsThresholdCount) {
Map<String, String> compactionConfigs = new HashMap<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
