Refactor TSDataType switches to use TypeService - #18606
Conversation
| public GroupedAvgAccumulator(TSDataType argumentDataType) { | ||
| this.argumentDataType = argumentDataType; | ||
| Type type = Type.fromTsDataType(argumentDataType); | ||
| this.inputStrategy = TypeServices.GROUPED_AVG_INPUT_SERVICE.call(type); |
There was a problem hiding this comment.
Use a typed batch loop for grouped AVG so input type dispatch is outside the row loop. Group IDs still use logical input positions, and count/sum updates preserve input order. NumericBatchAggregationTest verifies masks, group placement and floating-point order; local mixed-type kernels improved 3.08–3.38x, with fixed-DOUBLE performance flat or better.
| this.minResult = TsPrimitiveType.getByType(seriesDataType); | ||
| this.type = Type.fromTsDataType(seriesDataType); | ||
| this.minResult = type.getTsPrimitiveType(); | ||
| this.valueUpdater = TypeServices.MIN_COLUMN_BATCH_UPDATER_SERVICE.call(type); |
There was a problem hiding this comment.
MIN now selects a TypeService strategy containing the native scan, removing the shared per-value updater dispatch. NumericBatchAggregationTest verifies masked/null inputs and the existing NaN, signed-zero and integer-boundary behavior. The local mixed-type benchmark improved; fixed-DOUBLE results were smaller or approximately flat.
| this.maxResult = TsPrimitiveType.getByType(seriesDataType); | ||
| this.type = Type.fromTsDataType(seriesDataType); | ||
| this.maxResult = type.getTsPrimitiveType(); | ||
| this.valueUpdater = TypeServices.MAX_COLUMN_BATCH_UPDATER_SERVICE.call(type); |
There was a problem hiding this comment.
MAX now selects a TypeService strategy containing the native scan, removing the shared per-value updater dispatch. NumericBatchAggregationTest verifies masked/null inputs and the existing NaN, signed-zero and integer-boundary behavior. The local mixed-type benchmark improved; fixed-DOUBLE results were smaller or approximately flat.
| this.extremeResult = TsPrimitiveType.getByType(seriesDataType); | ||
| this.type = Type.fromTsDataType(seriesDataType); | ||
| this.extremeResult = type.getTsPrimitiveType(); | ||
| this.valueUpdater = TypeServices.EXTREME_COLUMN_BATCH_UPDATER_SERVICE.call(type); |
There was a problem hiding this comment.
EXTREME now selects a TypeService strategy containing the native scan, removing the shared per-value updater dispatch. NumericBatchAggregationTest verifies masked/null inputs and the existing NaN, signed-zero and integer-boundary behavior. The local mixed-type benchmark improved; fixed-DOUBLE results were smaller or approximately flat.
| TSDataType seriesDataType, VarianceAccumulator.VarianceType varianceType) { | ||
| this.seriesDataType = seriesDataType; | ||
| this.doubleValueConverter = | ||
| TypeServices.NUMERIC_BATCH_READER_SERVICE |
There was a problem hiding this comment.
Use the native numeric batch reader for grouped variance, retaining the existing recurrence callback and logical row order rather than changing the numerical algorithm. NumericMomentBatchTest covers moments/variance across tree, table and grouped implementations, including masked/null inputs. This candidate was retained after mixed-type and fixed-DOUBLE measurements; related candidates with measured regressions were restored.
| TSDataType seriesDataType, CentralMomentAccumulator.MomentType momentType) { | ||
| this.seriesDataType = seriesDataType; | ||
| this.doubleValueConverter = | ||
| TypeServices.NUMERIC_BATCH_READER_SERVICE |
There was a problem hiding this comment.
Use the native numeric batch reader for table central moment, retaining the existing recurrence callback and logical row order rather than changing the numerical algorithm. NumericMomentBatchTest covers moments/variance across tree, table and grouped implementations, including masked/null inputs. This candidate was retained after mixed-type and fixed-DOUBLE measurements; related candidates with measured regressions were restored.
| initializeOrValidateWindow(groupId, currentWindowStart, currentWindowEnd); | ||
| samples.add(groupId, time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
GroupedNaiveDeltaAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(groupId, currentWindowStart, currentWindowEnd); | ||
| samples.add(groupId, time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
GroupedNaiveIncreaseAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| arguments[1], position, RateFunctionType.IRATE, 2); | ||
| samples.add(groupId, time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
GroupedNaiveIrateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(groupId, currentWindowStart, currentWindowEnd); | ||
| samples.add(groupId, time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
GroupedNaiveRateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| arguments[1], position, RateFunctionType.IRATE, 2); | ||
| update(groupId, time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
GroupedOrderedIrateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| samples.add(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
NaiveDeltaAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| samples.add(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
NaiveIncreaseAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| arguments[1], position, RateFunctionType.IRATE, 2); | ||
| samples.add(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
NaiveIrateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| samples.add(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
NaiveRateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| update(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
OrderedDeltaAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| update(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
OrderedIncreaseAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| initializeOrValidateWindow(currentWindowStart, currentWindowEnd); | ||
| update(time, value); | ||
| } | ||
| TypeServices.RATE_INPUT_SERVICE |
There was a problem hiding this comment.
OrderedRateAccumulator now dispatches to a native batch reader through RATE_INPUT_SERVICE. It preserves value validation before time/window validation and keeps the existing state update callback, including logical group positions where applicable. RateBatchInputTest exercises all 16 implementations with native and wrapped columns, masks/nulls and validation cases. This is one of the 12 retained rate candidates; the other four were restored after fixed-DOUBLE benchmarks showed regressions.
| return value; | ||
| } | ||
|
|
||
| public static void validateValue(double value, RateFunctionType functionType) { |
There was a problem hiding this comment.
Expose the existing finite/nonnegative value checks to the typed batch readers without changing exception order or messages. Keep readValue’s scalar checks inline: replacing them with a helper changed the scalar benchmark behavior on paths whose batch refactor was rejected. RateBatchInputTest covers these validation semantics.
| Type returnType, ColumnTransformer childColumnTransformer, ZoneId zoneId) { | ||
| super(returnType, childColumnTransformer); | ||
| this.zoneId = zoneId; | ||
| this.numericCastBatch = |
There was a problem hiding this comment.
Select the numeric source/target batch strategy once for CAST/TRY_CAST. The 16 INT32/INT64/FLOAT/DOUBLE pairs use native loops; other types keep the existing conversion path. NumericCastBatchTest covers overflow, NaN, null/selection, region and RLE inputs, including TRY_CAST null results. Local mixed-type kernels improved 2.11–2.83x; fixed DOUBLE-to-DOUBLE was mostly smaller gains or flat.
| public VarianceAccumulator(TSDataType seriesDataType, VarianceType varianceType) { | ||
| this.seriesDataType = seriesDataType; | ||
| this.doubleValueConverter = | ||
| TypeServices.NUMERIC_BATCH_READER_SERVICE |
There was a problem hiding this comment.
Use the native numeric batch reader for tree variance, retaining the existing recurrence callback and logical row order rather than changing the numerical algorithm. NumericMomentBatchTest covers moments/variance across tree, table and grouped implementations, including masked/null inputs. This candidate was retained after mixed-type and fixed-DOUBLE measurements; related candidates with measured regressions were restored.
| value -> { | ||
| throw new UnsupportedOperationException( | ||
| String.format( | ||
| "The type %s cannot be casted to int.", type.getTypeEnum())); |
There was a problem hiding this comment.
[P2] Preserve localized errors in the extracted type services
This branch replaces DataNodePipeMessages.PIPE_EXCEPTION_THE_TYPE_S_CANNOT_BE_CASTED_TO_INT_659069CC with a raw English string. In a -P with-zh-locale build, calling CustomizedReadableIntermediateResults.getInt() for a BOOLEAN intermediate result now emits The type BOOLEAN cannot be casted to int. instead of the existing Chinese message.
Please reuse the existing DataNodePipeMessages constants for these conversion errors. The long/float/double/string strategies below have the same regression. The extracted ROUND_TRANSFORMER_SERVICE, unsupportedSourceCastStrategy, and unsupportedTargetCastStrategy also replace DataNodeQueryMessages constants with English literals; please restore those localized messages as well.
jt2594838
left a comment
There was a problem hiding this comment.
Merged master at 9be80ea and resolved the OPC UA migration compatibility issues in 5fdbfc7. The following inline notes cover the registry cleanup, type-service relocation, and regression-test relocation. Validation: 66 targeted tests passed; English and Chinese test-compile passed for all 52 modules excluding distribution; Spotless, Checkstyle, and git diff --check passed. Distribution packaging was not validated: the initial full-reactor command stalled while downloading a remote distribution ZIP, so compilation was rerun excluding that packaging module.
| @@ -157,7 +157,6 @@ public String getClassName() { | |||
| IOT_CONSENSUS_V2_ASYNC_CONNECTOR.getPipePluginName().toLowerCase(), | |||
| PIPE_CONSENSUS_ASYNC_CONNECTOR.getPipePluginName().toLowerCase(), | |||
| WEBSOCKET_CONNECTOR.getPipePluginName().toLowerCase(), | |||
| OPC_UA_CONNECTOR.getPipePluginName().toLowerCase(), | |||
There was a problem hiding this comment.
Remove the stale OPC_UA_CONNECTOR and OPC_UA_SINK entries from BUILTIN_SINKS. Master moved OPC UA to library-pipe and removed those enum constants, but these two references remained, causing node-commons compilation to fail. Removing both entries also correctly classifies the separately registered OPC UA plugin as external. The compatibility names in PipeSinkConstant remain available for configuration and the single-thread policy. Both locale builds passed after this cleanup.
| import java.util.function.Supplier; | ||
|
|
||
| // Keep Milo-dependent type services in the external plugin so core can load without Milo. | ||
| final class OpcUaTypeServices { |
There was a problem hiding this comment.
Relocate the four OPC UA TypeServices from datanode's TypeServices.Pipe into the external plugin, and update all four OpcUaNameSpace call sites to use this class. Master removed Milo from datanode's dependencies; retaining DateTime/NodeId/Identifiers references there prevented compilation. Keeping the services with the plugin preserves the existing conversions, service checks, and Pub/Sub type selection outside the row loop without reintroducing Milo into core. The java.util.Date conversion fix is retained. Both locale reactor builds and the targeted OPC UA metadata/conversion tests passed.
|
|
||
| public class OpcUaTypeServicesTest { | ||
|
|
||
| @Test |
There was a problem hiding this comment.
Move the existing DATE regression from datanode's TypeServicesTest alongside the relocated OPC UA services. The same assertion still checks Milo's DateTime against the LocalDate at the system zone's start of day, guarding against reintroducing java.sql.Date.toInstant(), which throws UnsupportedOperationException. This keeps regression coverage while allowing datanode tests to compile without Milo. The relocated test passed, together with five OPC UA metadata tests and the remaining 60 targeted tests.
What is changed
2.4.1-260915-SNAPSHOT.OrderedIrateAccumulator,GroupedOrderedRateAccumulator,GroupedOrderedIncreaseAccumulator, andGroupedOrderedDeltaAccumulatorinto the existingRATE_INPUT_SERVICEtype-specific loops. Type dispatch occurs once per batch, while null handling, logical mask positions, value/time/window validation, and ordered state updates retain their original semantics.AlignedTVListWAL writes by column and physical array segment throughALIGNED_WAL_BATCH_WRITER_SERVICE. Each segment is cast once, preserving value/null-flag interleaving and placeholders for unmaterialized segments.GroupedApproxCountDistinctAccumulatorsparse-mask traversal to use the selected-position count, preventing out-of-bounds access and processing of unselected rows.AlignedTVList: include the existing payload in the total size and read the bitmap length as the fixed-width integer already emitted by the WAL writer. The serialized format is unchanged.Performance measurements
Standalone JVM microbenchmarks compare an independently compiled baseline at
c4ec742b2ebc29ba53bd0ce1c96d9a990663fb9dwith the retained optimization patch. The independent HLL mask and WAL bitmap correctness fixes were applied to both sides so they execute equivalent workloads. These measurements isolate the optimization experiment and are not a measurement of the intervening master merge.Ordered rate accumulators
OrderedIrateAccumulatorGroupedOrderedRateAccumulatorGroupedOrderedIncreaseAccumulatorGroupedOrderedDeltaAccumulatorFixed-type ranges cover separate INT32, INT64, FLOAT, and DOUBLE runs. All 240 rate measurements produced matching baseline/candidate checksums.
Tradeoff:
OrderedIrateAccumulatoris approximately 3%–9% slower for fixed-type, all-row input, adding 0.038–0.116 ns per row. It is retained for the substantially larger mixed-type and sparse-mask gains; this is not a universal speedup.Aligned WAL writes
Other repositories were building concurrently on this machine, and storage samples showed substantial variability. These are local microbenchmark observations, not end-to-end IoTDB throughput estimates; small differences should be treated as noise. The TsFile JAR checksum remained unchanged throughout the final measurements.
The additional shared grouped/min-by, Percentile/HLL batch-input, IN/NOT IN, WAL batch-reader, and MemPageReader candidates were excluded after testing because their gains did not generalize or some configurations regressed. Their measurements are not included as improvements above.
Validation
For the additional optimization and correctness patch:
-pl iotdb-core/datanode -am).git diff --checkpassed.Earlier PR validation also passed the ArchUnit TSDataType switch checks without a baseline and REST Chinese-locale test compilation.