This is an automated email from the ASF dual-hosted git repository.
9aman 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 1d0aae09627 Reject time-based flush threshold for pauseless tables
(#20168)
1d0aae09627 is described below
commit 1d0aae096273782890db7e02b731b45063d2bdb4
Author: 9aman <[email protected]>
AuthorDate: Tue Nov 10 15:27:26 2026 +0530
Reject time-based flush threshold for pauseless tables (#20168)
---
.../segment/local/utils/TableConfigUtils.java | 30 +++++++++++++++-
.../segment/local/utils/TableConfigUtilsTest.java | 40 ++++++++++++++++++++++
2 files changed, 69 insertions(+), 1 deletion(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index f3b9fc3a809..242278577e8 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -48,6 +48,7 @@ import
org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.FunctionContext;
import org.apache.pinot.common.request.context.RequestContextUtils;
import org.apache.pinot.common.tier.TierFactory;
+import org.apache.pinot.common.utils.PauselessConsumptionUtils;
import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.segment.local.aggregator.ValueAggregator;
import org.apache.pinot.segment.local.aggregator.ValueAggregatorFactory;
@@ -527,13 +528,26 @@ public final class TableConfigUtils {
||
CommonConstants.HTTPS_PROTOCOL.equalsIgnoreCase(peerSegmentDownloadScheme);
}
+ /// Returns whether the stored table config is already pauseless and sets
`key` to `value` in one of its stream
+ /// configs. This value stays valid, so that updates to an existing table
and to its schema keep working.
+ private static boolean hasSamePauselessTimeFlushThreshold(@Nullable
TableConfig existingTableConfig, String key,
+ String value) {
+ if (existingTableConfig == null ||
!PauselessConsumptionUtils.isPauselessEnabled(existingTableConfig)) {
+ return false;
+ }
+ // Pauseless consumption is enabled in the stream ingestion config, so the
ingestion config is not null
+ return
existingTableConfig.getIngestionConfig().getStreamIngestionConfig().getStreamConfigMaps().stream()
+ .anyMatch(streamConfigMap -> value.equals(streamConfigMap.get(key)));
+ }
+
/// Validates the table's [IngestionConfig] together with the closely
related metrics-aggregation config, covering:
/// - Metrics aggregation (both the `aggregateMetrics` flag and ingestion
`aggregationConfigs`), delegated to
/// [#validateMetricsAggregation].
/// - Batch ingestion: each batch config map is well-formed, and a dimension
table has batch ingestion configured
/// with `REFRESH` segment ingestion type.
/// - Stream ingestion: streams are declared in only one place, at least one
stream is present, and pauseless
- /// consumption has a valid `peerSegmentDownloadScheme`.
+ /// consumption has a valid `peerSegmentDownloadScheme` and no time flush
threshold, except the unchanged value of
+ /// a stored pauseless table.
/// - Filter config: the filter function is valid and not a disabled Groovy
expression.
/// - Source field configs: no source field is duplicated within the same
complex-type phase.
/// - Enrichment configs: each config is valid.
@@ -608,6 +622,20 @@ public final class TableConfigUtils {
} else {
LOGGER.warn("It's not recommended to create pauseless tables with
replication 1 for stability reasons.");
}
+ // Each replica checks a time flush threshold on its own clock, and a
replica that restarts near the end time
+ // extends it. So the replicas of a pauseless segment can stop at
different offsets. A replica past the
+ // committed end offset can only download the segment, so if the
committer fails before the upload, no replica
+ // has a copy. Reject an explicit time threshold (the default
threshold still applies). A stored pauseless
+ // table keeps its unchanged value, so that updates to the table and
to its schema keep working.
+ for (Map<String, String> streamConfigMap : streamConfigMaps) {
+ for (String key :
List.of(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME,
+ StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME +
StreamConfigProperties.LLC_SUFFIX)) {
+ String value = streamConfigMap.get(key);
+ Preconditions.checkState(
+ value == null ||
hasSamePauselessTimeFlushThreshold(existingTableConfig, key, value),
+ "%s is not supported with pauseless consumption enabled, use a
row or size based flush threshold", key);
+ }
+ }
}
}
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 6deb0aca427..9a952662bf6 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
@@ -1201,6 +1201,46 @@ public class TableConfigUtilsTest {
}
}
+ @Test
+ public void pauselessTimeFlushThresholdTest() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addDateTime("timeColumn", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ for (String key :
List.of(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME,
+ StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME +
StreamConfigProperties.LLC_SUFFIX)) {
+ TableConfig tableConfig = getTimeFlushThresholdTableConfig(key, "30m",
true);
+ IllegalStateException e =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(tableConfig, schema));
+ assertTrue(e.getMessage().contains(key), e.getMessage());
+
+ // The same config is valid when pauseless consumption is disabled
+ TableConfig nonPauselessTableConfig =
getTimeFlushThresholdTableConfig(key, "30m", false);
+ TableConfigUtils.validate(nonPauselessTableConfig, schema);
+
+ // A stored pauseless table keeps its value, which also lets a schema
update revalidate it. A changed value, or
+ // enabling pauseless consumption on a table with the key, is rejected.
+ TableConfigUtils.validate(tableConfig, schema, null, tableConfig);
+ TableConfig changedTableConfig = getTimeFlushThresholdTableConfig(key,
"5m", true);
+ expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(changedTableConfig, schema, null,
tableConfig));
+ expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(tableConfig, schema, null,
nonPauselessTableConfig));
+ }
+ }
+
+ private TableConfig getTimeFlushThresholdTableConfig(String key, String
value, boolean pauselessEnabled) {
+ Map<String, String> streamConfigs = getStreamConfigs();
+ streamConfigs.put(key, value);
+ StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(streamConfigs));
+ streamIngestionConfig.setPauselessConsumptionEnabled(pauselessEnabled);
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
+ return new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(ingestionConfig)
+ .build();
+ }
+
@Test
public void ingestionServerIngestionOomProtectionTest() {
Schema schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).build();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
