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 16036c67aed Revert "Reject time-based flush threshold for pauseless
tables (#20168)" (#20221)
16036c67aed is described below
commit 16036c67aed73fca570986378b616cc598295e9d
Author: 9aman <[email protected]>
AuthorDate: Wed Nov 11 14:54:20 2026 +0530
Revert "Reject time-based flush threshold for pauseless tables (#20168)"
(#20221)
This reverts commit 1d0aae096273782890db7e02b731b45063d2bdb4.
Pauseless tables can set realtime.segment.flush.threshold.time and its
legacy .llc key again.
---
.../segment/local/utils/TableConfigUtils.java | 30 +---------------
.../segment/local/utils/TableConfigUtilsTest.java | 40 ----------------------
2 files changed, 1 insertion(+), 69 deletions(-)
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 242278577e8..f3b9fc3a809 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,7 +48,6 @@ 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;
@@ -528,26 +527,13 @@ 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` and no time flush
threshold, except the unchanged value of
- /// a stored pauseless table.
+ /// consumption has a valid `peerSegmentDownloadScheme`.
/// - 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.
@@ -622,20 +608,6 @@ 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 9a952662bf6..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
@@ -1201,46 +1201,6 @@ 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]
