This is an automated email from the ASF dual-hosted git repository.
KKcorps 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 3399014aa4e Cover configuration validation and FLOAT segment
preservation (#20151)
3399014aa4e is described below
commit 3399014aa4eb679403e1c520e680bb1b71ec77f6
Author: Kartik Khare <[email protected]>
AuthorDate: Tue Nov 10 14:17:46 2026 +0530
Cover configuration validation and FLOAT segment preservation (#20151)
* Cover policy validation and float preservation to expose boundary
regressions
* Remove JSON codec coverage to keep the change within scope
---------
Co-authored-by: Kartik Khare <[email protected]>
---
.../converter/RealtimeSegmentConverterTest.java | 89 ++++++++++++++++++++++
.../segment/local/utils/TableConfigUtilsTest.java | 32 ++++++++
2 files changed, 121 insertions(+)
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java
index 6d365817664..a3e148fef3b 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java
@@ -35,7 +35,9 @@ import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
+import
org.apache.pinot.segment.local.columntransformer.NullValueColumnTransformer;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl;
import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager;
import org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentConfig;
@@ -48,6 +50,7 @@ import
org.apache.pinot.segment.local.segment.index.text.TextIndexConfigBuilder;
import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
import
org.apache.pinot.segment.local.segment.virtualcolumn.VirtualColumnProviderFactory;
import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
@@ -468,6 +471,92 @@ public class RealtimeSegmentConverterTest implements
PinotBuffersAfterMethodChec
}
}
+ @DataProvider
+ public Object[][] mvFloatDefaults() {
+ List<Object[]> cases = new ArrayList<>();
+ for (boolean columnMajor : new boolean[]{false, true}) {
+ for (boolean raw : new boolean[]{false, true}) {
+ for (boolean finiteDefault : new boolean[]{false, true}) {
+ for (String input : List.of("omitted", "infinity", "finite")) {
+ for (int rows : new int[]{1, 257}) {
+ cases.add(new Object[]{columnMajor, raw, finiteDefault, input,
rows});
+ }
+ }
+ }
+ }
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ /// Ordinary conversion and reopen must preserve the schema default and
explicit FLOAT values without any upsert.
+ @Test(dataProvider = "mvFloatDefaults")
+ public void testMvFloatDefaultsSurviveOrdinarySeal(boolean columnMajor,
boolean raw, boolean finiteDefault,
+ String input, int rows)
+ throws Exception {
+ String caseName = "floatDefaults-" + columnMajor + "-" + raw + "-" +
finiteDefault + "-" + input + "-" + rows;
+ File tmp = new File(TMP_DIR, caseName);
+ FileUtils.forceMkdir(tmp);
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("floatDefaults")
+ .addMultiValueDimension("v", DataType.FLOAT)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS").build();
+ if (finiteDefault) {
+ schema.getFieldSpecFor("v").setDefaultNullValue(7.25f);
+ }
+ TableConfig table = new
TableConfigBuilder(TableType.REALTIME).setTableName("floatDefaults")
+
.setTimeColumnName("ts").setColumnMajorSegmentBuilderEnabled(columnMajor)
+ .setNoDictionaryColumns(raw ? List.of("v") : List.of()).build();
+ String segmentName = "floatDefaults__0__0__123456";
+ RealtimeSegmentConfig.Builder builder = new RealtimeSegmentConfig.Builder()
+
.setTableNameWithType(table.getTableName()).setSegmentName(segmentName).setStreamName(table.getTableName())
+
.setSchema(schema).setTimeColumnName("ts").setCapacity(1000).setAvgNumMultiValues(2)
+
.setSegmentZKMetadata(getSegmentZKMetadata(segmentName)).setOffHeap(true)
+ .setMemoryManager(new DirectMemoryManager(segmentName))
+ .setStatsHistory(RealtimeSegmentStatsHistory.deserializeFrom(new
File(tmp, "stats")))
+ .setConsumerDir(new File(tmp, "consumer").getAbsolutePath());
+ if (raw) {
+ builder.setIndex(Set.of("v"), StandardIndexes.dictionary(),
DictionaryIndexConfig.DISABLED)
+ .setIndex(Set.of("v"), StandardIndexes.forward(),
+ ForwardIndexConfig.getDefault(FieldConfig.EncodingType.RAW));
+ }
+ MutableSegmentImpl mutable = new MutableSegmentImpl(builder.build(), null);
+ try {
+ float expected = input.equals("finite")
+ ? 3.5f
+ : input.equals("omitted") && finiteDefault ? 7.25f :
Float.NEGATIVE_INFINITY;
+ GenericRow row = new GenericRow();
+ row.putValue("ts", 1793750400000L);
+ if (input.equals("omitted")) {
+ Object defaultValue =
+ new NullValueColumnTransformer(table, schema.getFieldSpecFor("v"),
schema).transform(null);
+ row.putDefaultNullValue("v", defaultValue);
+ } else {
+ row.putValue("v", new Object[]{expected});
+ }
+ for (int i = 0; i < rows; i++) {
+ mutable.index(row, null);
+ }
+ GenericRow before = mutable.getRecord(rows - 1, new GenericRow());
+ assertEquals(((Number) ((Object[])
before.getValue("v"))[0]).floatValue(), expected, "mutable " + caseName);
+ File output = new File(tmp, "output");
+ RealtimeSegmentConverter converter = new
RealtimeSegmentConverter(mutable, new SegmentZKPropsConfig(),
+ output.getAbsolutePath(), schema, table.getTableName(), table,
segmentName, false);
+ converter.build(SegmentVersion.v3);
+ for (int reopen = 0; reopen < 2; reopen++) {
+ ImmutableSegment immutable = ImmutableSegmentLoader.load(new
File(output, segmentName),
+ new IndexLoadingConfig(table, schema));
+ try {
+ GenericRow after = immutable.getRecord(rows - 1, new GenericRow());
+ float actual = ((Number) ((Object[])
after.getValue("v"))[0]).floatValue();
+ assertEquals(actual, expected, "ordinary seal/reopen " + reopen + "
" + caseName);
+ } finally {
+ immutable.destroy();
+ }
+ }
+ } finally {
+ mutable.destroy();
+ }
+ }
+
private void testSegment(List<GenericRow> rows, File indexDir,
TableConfig tableConfig, SegmentMetadataImpl segmentMetadata)
throws IOException, ConfigurationException {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 73edeac2f18..6deb0aca427 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -2858,6 +2858,30 @@ public class TableConfigUtilsTest {
}
}
+ @Test
+ public void testConsumptionPolicyMatrix() {
+ ConsumingSegmentConsistencyModeListener listener =
ConsumingSegmentConsistencyModeListener.getInstance();
+ try {
+ for (boolean protectedMode : new boolean[]{false, true}) {
+ listener.setMode(protectedMode ?
ConsumingSegmentConsistencyModeListener.Mode.PROTECTED
+ : ConsumingSegmentConsistencyModeListener.Mode.RESTRICTED);
+ for (UpsertConfig.Mode mode : new
UpsertConfig.Mode[]{UpsertConfig.Mode.FULL, UpsertConfig.Mode.PARTIAL}) {
+ for (boolean drop : new boolean[]{false, true}) {
+ UpsertConfig config = new UpsertConfig(mode);
+ config.setDropOutOfOrderRecord(drop);
+ for (ParallelSegmentConsumptionPolicy policy :
ParallelSegmentConsumptionPolicy.values()) {
+ boolean rejects = protectedMode && (mode ==
UpsertConfig.Mode.PARTIAL || drop)
+ && policy !=
ParallelSegmentConsumptionPolicy.DISALLOW_ALWAYS;
+ checkConsumptionDuringUpsertRevert(config, policy, rejects);
+ }
+ }
+ }
+ }
+ } finally {
+ listener.reset();
+ }
+ }
+
private void checkConsumptionDuringUpsertRevert(UpsertConfig upsertConfig,
ParallelSegmentConsumptionPolicy policy,
boolean expectRejected) {
TableConfig tableConfig =
createTableConfigWithConsumptionPolicy(upsertConfig, policy);
@@ -4872,6 +4896,14 @@ public class TableConfigUtilsTest {
assertNull(TableConfigUtils.getPartitionColumn(tableConfig));
}
+ @Test
+ public void testUnknownSkipTokenCannotBypassValidation() {
+ for (String value : List.of("ACTIVE_TASKS", "TASK,ACTIVE_TASKS",
"ALL,ACTIVE_TASKS", "active_tasks")) {
+ expectThrows(IllegalArgumentException.class, () ->
TableConfigUtils.parseTypesToSkipString(value));
+ }
+ assertEquals(TableConfigUtils.parseTypesToSkipString(" task , TASK "),
EnumSet.of(ValidationType.TASK));
+ }
+
@Test
public void testParseTypesToSkipString() {
assertNull(TableConfigUtils.parseTypesToSkipString(null));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
