From 4ff94896ea602704d19226dd34f0ad4079c58528 Mon Sep 17 00:00:00 2001 From: Dmitry Leontyev Date: Tue, 24 Feb 2026 12:12:16 +0100 Subject: [PATCH 1/5] MetricsInfoOperator rafctoring Make constructors private. Use scratch instead of creating BytesRef each time we need to read _source. --- .../compute/operator/MetricsInfoOperator.java | 11 +- .../operator/MetricsInfoOperatorTests.java | 138 +++++++++++------- 2 files changed, 90 insertions(+), 59 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java index bc65d09debb8a..904c2b64a2a45 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java @@ -199,7 +199,7 @@ public enum Mode { /** * Creates an INITIAL-mode operator (data nodes). */ - public MetricsInfoOperator(BlockFactory blockFactory, MetricFieldLookup fieldLookup, int metadataSourceChannel, int indexChannel) { + private MetricsInfoOperator(BlockFactory blockFactory, MetricFieldLookup fieldLookup, int metadataSourceChannel, int indexChannel) { this.mode = Mode.INITIAL; this.blockFactory = blockFactory; this.breaker = blockFactory.breaker(); @@ -215,7 +215,7 @@ public MetricsInfoOperator(BlockFactory blockFactory, MetricFieldLookup fieldLoo * @param channels the 6 input channel indices mapping to * [metric_name, data_stream, unit, metric_type, field_type, dimension_fields] */ - public MetricsInfoOperator(BlockFactory blockFactory, int[] channels) { + private MetricsInfoOperator(BlockFactory blockFactory, int[] channels) { this.mode = Mode.FINAL; this.blockFactory = blockFactory; this.breaker = blockFactory.breaker(); @@ -246,6 +246,7 @@ private void addInputInitial(Page page) { BytesRefBlock indexBlock = page.getBlock(indexChannel); BytesRef indexScratch = new BytesRef(); + BytesRef sourceScratch = new BytesRef(); for (int p = 0; p < page.getPositionCount(); p++) { if (metadataSource.isNull(p)) { @@ -257,7 +258,7 @@ private void addInputInitial(Page page) { String indexName = indexBlock.getBytesRef(p, indexScratch).utf8ToString(); String dataStreamName = resolveDataStreamName(indexName); - Map metadata = parseMetadataSource(metadataSource, p); + Map metadata = parseMetadataSource(metadataSource, p, sourceScratch); if (metadata == null) { continue; } @@ -530,11 +531,11 @@ private static void appendMultiValued(BytesRefBlock.Builder builder, Set } } - private Map parseMetadataSource(BytesRefBlock metadataSource, int position) { + private Map parseMetadataSource(BytesRefBlock metadataSource, int position, BytesRef sourceScratch) { if (metadataSource == null || metadataSource.isNull(position)) { return null; } - BytesRef bytes = metadataSource.getBytesRef(position, new BytesRef()); + BytesRef bytes = metadataSource.getBytesRef(position, sourceScratch); try ( var parser = XContentType.JSON.xContent() .createParser(XContentParserConfiguration.EMPTY, bytes.bytes, bytes.offset, bytes.length) diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java index d65121177fa5a..021f7d9bedd84 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java @@ -110,9 +110,29 @@ protected void assertStatus(@Nullable Map map, List input, assertNull(map); } + private MetricsInfoOperator createInitialOperator() { + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL).get(driverContext()); + } + + private MetricsInfoOperator createInitialOperator(MetricsInfoOperator.MetricFieldLookup lookup) { + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(driverContext()); + } + + private MetricsInfoOperator createInitialOperator( + MetricsInfoOperator.MetricFieldLookup lookup, + int metadataSourceChannel, + int indexChannel + ) { + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, metadataSourceChannel, indexChannel).get(driverContext()); + } + + private MetricsInfoOperator createFinalOperator(int[] channels) { + return (MetricsInfoOperator) new MetricsInfoOperator.FinalFactory(channels).get(driverContext()); + } + public void testSingleMetricSingleIndex() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.85, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -135,7 +155,7 @@ public void testSingleMetricSingleIndex() { public void testMultipleMetricsSameIndex() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Both cpu_usage and disk_io are metric fields Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"disk_io\": 1024, \"host\": \"server1\"}", "my-index"); op.addInput(input); @@ -154,7 +174,7 @@ public void testMultipleMetricsSameIndex() { public void testSameMetricDifferentIndicesMergedBySignature() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Same metric from two different indices but with the same signature → merged into one row Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "index-a"); op.addInput(input1); @@ -180,9 +200,10 @@ public void testSameMetricDifferentIndicesMergedBySignature() { } public void testDifferentSignaturesNotMerged() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); // Lookup returns different types for the same metric name in different indices - try (MetricsInfoOperator op = getMetricsInfoOperator(blockFactory)) { + try (MetricsInfoOperator op = getMetricsInfoOperator(ctx)) { Page input1 = buildPage(blockFactory, "{\"cpu\": 0.5}", "index-a"); op.addInput(input1); @@ -200,7 +221,7 @@ public void testDifferentSignaturesNotMerged() { } } - private static MetricsInfoOperator getMetricsInfoOperator(BlockFactory blockFactory) { + private static MetricsInfoOperator getMetricsInfoOperator(DriverContext ctx) { MetricsInfoOperator.MetricFieldLookup lookup = (indexName, fieldName) -> { if ("cpu".equals(fieldName)) { if ("index-a".equals(indexName)) { @@ -211,13 +232,12 @@ private static MetricsInfoOperator getMetricsInfoOperator(BlockFactory blockFact } return null; }; - - return new MetricsInfoOperator(blockFactory, lookup, METADATA_CHANNEL, INDEX_CHANNEL); + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(ctx); } public void testDimensionFieldsCollected() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // "host" and "region" are non-metric (dimension) fields Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\", \"region\": \"us-east\"}", "my-index"); op.addInput(input); @@ -244,7 +264,7 @@ public void testNestedMetadataFields() { return null; }; - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, lookup, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(lookup)) { Page input = buildPage(blockFactory, "{\"system\": {\"cpu\": 0.75}, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -268,7 +288,7 @@ public void testNestedMetadataFields() { */ public void testDimensionFieldsUnderPassthroughStylePath() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Metadata shaped like a passthrough: resource.attributes.service.name (dimension), cpu_usage (metric) String metadataJson = """ {"resource": {"attributes": {"service": {"name": "api"}}}, "cpu_usage": 0.5} @@ -296,7 +316,7 @@ public void testDimensionFieldsUnderPassthroughStylePath() { */ public void testBothTopLevelAndPassthroughDimensionPaths() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Two documents: one with top-level service.name, one with resource.attributes.service.name Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"service\": {\"name\": \"web\"}}", "my-index"); Page input2 = buildPage( @@ -321,7 +341,7 @@ public void testBothTopLevelAndPassthroughDimensionPaths() { public void testNullMetadataSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -351,7 +371,7 @@ public void testNullMetadataSkipped() { public void testNullIndexSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -380,7 +400,7 @@ public void testNullIndexSkipped() { public void testEmptyInputProducesEmptyPage() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { op.finish(); Page output = op.getOutput(); @@ -394,7 +414,7 @@ public void testEmptyInputProducesEmptyPage() { public void testGetOutputBeforeFinishReturnsNull() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { assertNull(op.getOutput()); assertTrue(op.needsInput()); } @@ -402,7 +422,7 @@ public void testGetOutputBeforeFinishReturnsNull() { public void testGetOutputCalledTwiceReturnsNullSecondTime() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -419,7 +439,7 @@ public void testGetOutputCalledTwiceReturnsNullSecondTime() { public void testNeedsInputReturnsFalseAfterFinish() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { assertTrue(op.needsInput()); op.finish(); assertFalse(op.needsInput()); @@ -428,7 +448,7 @@ public void testNeedsInputReturnsFalseAfterFinish() { public void testIsFinishedOnlyAfterOutputProduced() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { assertFalse(op.isFinished()); op.finish(); @@ -452,7 +472,7 @@ public void testMetricWithNullUnit() { return null; }; - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, lookup, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(lookup)) { Page input = buildPage(blockFactory, "{\"temperature\": 36.6}", "my-index"); op.addInput(input); op.finish(); @@ -472,7 +492,7 @@ public void testMetricWithNullUnit() { public void testInvalidJsonSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -502,7 +522,7 @@ public void testInvalidJsonSkipped() { public void testMultipleInputPages() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Page 1: cpu_usage from index-a Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "index-a"); op.addInput(input1); @@ -525,7 +545,7 @@ public void testMultipleInputPages() { public void testDimensionFieldsAreUnionAcrossDocuments() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Document 1: dimension "host" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); op.addInput(input1); @@ -555,7 +575,7 @@ public void testFactoryDescribe() { public void testToString() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, 0, 1)) { + try (MetricsInfoOperator op = createInitialOperator(SIMPLE_LOOKUP, 0, 1)) { assertThat(op.toString(), equalTo("MetricsInfoOperator[mode=INITIAL]")); } } @@ -582,7 +602,7 @@ public void testResolveDataStreamName() { public void testDifferentMetricsHaveDifferentDimensions() { BlockFactory blockFactory = driverContext().blockFactory(); // Lookup: cpu_usage and disk_io are metrics - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Tsid 1: has cpu_usage + dimension "host" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); // Tsid 2: has disk_io + dimension "mount" @@ -620,7 +640,7 @@ public void testDifferentMetricsHaveDifferentDimensions() { */ public void testBackingIndicesSameDataStreamGroupedTogether() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Two backing indices of the same data stream "k8s" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu_usage\": 0.9, \"host\": \"b\"}", ".ds-k8s-2024.01.15-000002"); @@ -645,9 +665,10 @@ public void testBackingIndicesSameDataStreamGroupedTogether() { * same data stream have conflicting values. */ public void testConflictingValuesWithinSameDataStreamProduceMultiValuedFields() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); // Lookup returns different field_type depending on backing index - try (MetricsInfoOperator op = getInfoOperator(blockFactory)) { + try (MetricsInfoOperator op = getInfoOperator(ctx)) { // Two backing indices of the same data stream "k8s", different field_type Page input1 = buildPage(blockFactory, "{\"cpu\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu\": 0.9, \"host\": \"b\"}", ".ds-k8s-2024.01.15-000002"); @@ -668,7 +689,7 @@ public void testConflictingValuesWithinSameDataStreamProduceMultiValuedFields() } } - private static MetricsInfoOperator getInfoOperator(BlockFactory blockFactory) { + private static MetricsInfoOperator getInfoOperator(DriverContext ctx) { MetricsInfoOperator.MetricFieldLookup lookup = (indexName, fieldName) -> { if ("cpu".equals(fieldName)) { if (indexName.endsWith("-000001")) { @@ -679,7 +700,7 @@ private static MetricsInfoOperator getInfoOperator(BlockFactory blockFactory) { } return null; }; - return new MetricsInfoOperator(blockFactory, lookup, METADATA_CHANNEL, INDEX_CHANNEL); + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(ctx); } /** @@ -687,9 +708,10 @@ private static MetricsInfoOperator getInfoOperator(BlockFactory blockFactory) { * separate rows (one per data stream). */ public void testDifferentDataStreamsDifferentValuesProduceSeparateRows() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); // Lookup returns different unit depending on index - try (MetricsInfoOperator op = getOperator(blockFactory)) { + try (MetricsInfoOperator op = getOperator(ctx)) { Page input1 = buildPage(blockFactory, "{\"cpu\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu\": 0.9, \"host\": \"b\"}", ".ds-other-2024.01.15-000001"); op.addInput(input1); @@ -718,7 +740,7 @@ public void testDifferentDataStreamsDifferentValuesProduceSeparateRows() { } } - private static MetricsInfoOperator getOperator(BlockFactory blockFactory) { + private static MetricsInfoOperator getOperator(DriverContext ctx) { MetricsInfoOperator.MetricFieldLookup lookup = (indexName, fieldName) -> { if ("cpu".equals(fieldName)) { if (indexName.contains("k8s")) { @@ -729,7 +751,7 @@ private static MetricsInfoOperator getOperator(BlockFactory blockFactory) { } return null; }; - return new MetricsInfoOperator(blockFactory, lookup, METADATA_CHANNEL, INDEX_CHANNEL); + return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(ctx); } /** @@ -738,7 +760,7 @@ private static MetricsInfoOperator getOperator(BlockFactory blockFactory) { */ public void testDifferentDataStreamsSameValuesProduceMultiValuedDataStream() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator()) { // Two different data streams, same metric with the same signature Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu_usage\": 0.9, \"host\": \"b\"}", ".ds-other-2024.01.15-000001"); @@ -761,7 +783,7 @@ public void testDifferentDataStreamsSameValuesProduceMultiValuedDataStream() { public void testFinalModeIdenticalRowsFromTwoDataNodesAreMerged() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { // Two data nodes produced identical rows for cpu_usage on index-a Page page1 = buildFinalPage( blockFactory, @@ -799,7 +821,7 @@ public void testFinalModeIdenticalRowsFromTwoDataNodesAreMerged() { public void testFinalModePartiallyOverlappingDataStreamsAreMerged() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { // Data node 1 saw index-a and index-b; data node 2 saw index-a and index-c Page page1 = buildFinalPage( blockFactory, @@ -835,7 +857,7 @@ public void testFinalModePartiallyOverlappingDataStreamsAreMerged() { public void testFinalModeDifferentMetricNamesRemainSeparateRows() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { Page page1 = buildFinalPage( blockFactory, "cpu_usage", @@ -870,7 +892,7 @@ public void testFinalModeDifferentMetricNamesRemainSeparateRows() { public void testFinalModeDifferentSignaturesProduceSeparateRows() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { // Same metric name but different field_type → different signatures → 2 rows Page page1 = buildFinalPage( blockFactory, @@ -905,7 +927,7 @@ public void testFinalModeDifferentSignaturesProduceSeparateRows() { public void testFinalModeEmptyInputProducesEmptyOutput() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { op.finish(); Page output = op.getOutput(); @@ -919,7 +941,7 @@ public void testFinalModeEmptyInputProducesEmptyOutput() { public void testFinalModeSingleDataNodeInputPassesThrough() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { Page page = buildFinalPage( blockFactory, "cpu_usage", @@ -949,7 +971,7 @@ public void testFinalModeSingleDataNodeInputPassesThrough() { public void testFinalModeDimensionFieldsAreUnionedAcrossDataNodes() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { // Data node 1 saw dimension "host"; data node 2 saw dimension "region" Page page1 = buildFinalPage( blockFactory, @@ -991,7 +1013,7 @@ public void testFinalModeDimensionFieldsAreUnionedAcrossDataNodes() { */ public void testFinalModeSameDataStreamDifferentUnitsAcrossNodesProduceOneRow() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { // Data node A has backing index 000001 (unit=usd) Page page1 = buildFinalPage( blockFactory, @@ -1093,7 +1115,7 @@ private static void appendSet(BytesRefBlock.Builder builder, Set values) */ public void testHistogramMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, HISTOGRAM_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"legacy": {"values": [1.0, 2.0, 3.0], "counts": [10, 20, 30]}}, "entity": {"id": "abc"}}"""; Page input = buildPage(blockFactory, metadataJson.trim(), ".ds-histograms-2026.02.16-000001"); @@ -1123,7 +1145,7 @@ public void testHistogramMetricRecognizedFromNestedSyntheticSource() { */ public void testExponentialHistogramMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, HISTOGRAM_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"exponential": {"scale": 3, "sum": 42.5, "min": 1.0, "max": 10.0,\ "zero": {"count": 0, "threshold": 0.0},\ @@ -1156,7 +1178,7 @@ public void testExponentialHistogramMetricRecognizedFromNestedSyntheticSource() */ public void testTdigestMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, HISTOGRAM_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"tdigest": {"min": 0.5, "max": 99.5, "sum": 500.0,\ "centroids": [1.0, 50.0, 99.0], "counts": [100, 200, 100]}},\ @@ -1187,7 +1209,7 @@ public void testTdigestMetricRecognizedFromNestedSyntheticSource() { */ public void testMixedScalarAndHistogramMetrics() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, HISTOGRAM_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { // cpu_usage is a scalar metric (leaf), histogram.legacy is a complex metric (nested Map) String metadataJson = """ {"cpu_usage": 0.75, "histogram": {"legacy": {"values": [1.0], "counts": [5]}}, "host": "srv1"}"""; @@ -1220,7 +1242,7 @@ public void testMixedScalarAndHistogramMetrics() { */ public void testDimensionsNotContaminatedByHistogramInternalFields() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, HISTOGRAM_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { // All three histogram types in one document, plus a dimension String metadataJson = """ {"histogram": {\ @@ -1324,9 +1346,13 @@ private static Set collectMultiValues(Page page, int blockIndex, int pos } public void testInitialModeTracksMemoryOnNewEntries() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); long usedBefore = blockFactory.breaker().getUsed(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL)) { + try ( + MetricsInfoOperator op = (MetricsInfoOperator) new MetricsInfoOperator.Factory(SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL) + .get(ctx) + ) { Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"h1\"}", "index-a"); op.addInput(input1); long usedAfterOne = blockFactory.breaker().getUsed(); @@ -1352,9 +1378,10 @@ public void testInitialModeTracksMemoryOnNewEntries() { } public void testFinalModeTracksMemoryOnNewEntries() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); long usedBefore = blockFactory.breaker().getUsed(); - try (MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, FINAL_CHANNELS)) { + try (MetricsInfoOperator op = (MetricsInfoOperator) new MetricsInfoOperator.FinalFactory(FINAL_CHANNELS).get(ctx)) { Page page1 = buildFinalPage( blockFactory, "cpu_usage", @@ -1404,10 +1431,13 @@ public void testFinalModeTracksMemoryOnNewEntries() { } public void testCloseReleasesTrackedMemory() { - BlockFactory blockFactory = driverContext().blockFactory(); + DriverContext ctx = driverContext(); + BlockFactory blockFactory = ctx.blockFactory(); long usedBefore = blockFactory.breaker().getUsed(); - MetricsInfoOperator op = new MetricsInfoOperator(blockFactory, SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL); + MetricsInfoOperator op = (MetricsInfoOperator) new MetricsInfoOperator.Factory(SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL).get( + ctx + ); Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"h1\"}", "index-a"); op.addInput(input); assertThat(blockFactory.breaker().getUsed() - usedBefore, equalTo(MetricsInfoOperator.SHALLOW_SIZE)); From ba89a093c9d937522351f1ba871d090331b4e277 Mon Sep 17 00:00:00 2001 From: Dmitry Leontyev Date: Tue, 24 Feb 2026 12:18:21 +0100 Subject: [PATCH 2/5] Update docs/changelog/142935.yaml --- docs/changelog/142935.yaml | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 docs/changelog/142935.yaml diff --git a/docs/changelog/142935.yaml b/docs/changelog/142935.yaml new file mode 100644 index 0000000000000..048f62846acd7 --- /dev/null +++ b/docs/changelog/142935.yaml @@ -0,0 +1,5 @@ +area: "ES|QL, ES|QL" +issues: [] +pr: 142935 +summary: '`MetricsInfoOperator` rafctoring' +type: enhancement From 1a4eef0263a0a088c56adc5f1da6c33592412504 Mon Sep 17 00:00:00 2001 From: Dmitry Leontyev Date: Tue, 24 Feb 2026 12:19:08 +0100 Subject: [PATCH 3/5] MetricsInfoOperator rafctoring Make constructors private. Use scratch instead of creating BytesRef each time we need to read _source. --- docs/changelog/142935.yaml | 2 +- .../compute/operator/MetricsInfoOperatorTests.java | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/docs/changelog/142935.yaml b/docs/changelog/142935.yaml index 048f62846acd7..ea72f79107d6c 100644 --- a/docs/changelog/142935.yaml +++ b/docs/changelog/142935.yaml @@ -1,4 +1,4 @@ -area: "ES|QL, ES|QL" +area: ES|QL issues: [] pr: 142935 summary: '`MetricsInfoOperator` rafctoring' diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java index 021f7d9bedd84..c8727b9cb41bf 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java @@ -447,7 +447,6 @@ public void testNeedsInputReturnsFalseAfterFinish() { } public void testIsFinishedOnlyAfterOutputProduced() { - BlockFactory blockFactory = driverContext().blockFactory(); try (MetricsInfoOperator op = createInitialOperator()) { assertFalse(op.isFinished()); From 304e9d05ea9c14acf251c33f8ec1fc15a8815a99 Mon Sep 17 00:00:00 2001 From: Dmitry Leontyev Date: Tue, 24 Feb 2026 14:18:39 +0100 Subject: [PATCH 4/5] Delete docs/changelog/142935.yaml --- docs/changelog/142935.yaml | 5 ----- 1 file changed, 5 deletions(-) delete mode 100644 docs/changelog/142935.yaml diff --git a/docs/changelog/142935.yaml b/docs/changelog/142935.yaml deleted file mode 100644 index ea72f79107d6c..0000000000000 --- a/docs/changelog/142935.yaml +++ /dev/null @@ -1,5 +0,0 @@ -area: ES|QL -issues: [] -pr: 142935 -summary: '`MetricsInfoOperator` rafctoring' -type: enhancement From 33ded394194a7fa06e20394e522c95d50d664665 Mon Sep 17 00:00:00 2001 From: Dmitry Leontyev Date: Tue, 24 Feb 2026 16:20:27 +0100 Subject: [PATCH 5/5] Polish MetricsInfoOperator --- .../compute/operator/MetricsInfoOperator.java | 41 +++--- .../operator/MetricsInfoOperatorTests.java | 124 ++++++++---------- 2 files changed, 72 insertions(+), 93 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java index 904c2b64a2a45..98efdad9c5627 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/MetricsInfoOperator.java @@ -139,7 +139,14 @@ public interface MetricFieldLookup { public record Factory(MetricFieldLookup fieldLookup, int metadataSourceChannel, int indexChannel) implements OperatorFactory { @Override public Operator get(DriverContext driverContext) { - return new MetricsInfoOperator(driverContext.blockFactory(), fieldLookup, metadataSourceChannel, indexChannel); + return new MetricsInfoOperator( + Mode.INITIAL, + driverContext.blockFactory(), + fieldLookup, + metadataSourceChannel, + indexChannel, + null + ); } @Override @@ -161,7 +168,7 @@ public String describe() { public record FinalFactory(int[] channels) implements OperatorFactory { @Override public Operator get(DriverContext driverContext) { - return new MetricsInfoOperator(driverContext.blockFactory(), channels); + return new MetricsInfoOperator(Mode.FINAL, driverContext.blockFactory(), null, -1, -1, channels); } @Override @@ -196,32 +203,20 @@ public enum Mode { private boolean finished = false; private boolean outputProduced = false; - /** - * Creates an INITIAL-mode operator (data nodes). - */ - private MetricsInfoOperator(BlockFactory blockFactory, MetricFieldLookup fieldLookup, int metadataSourceChannel, int indexChannel) { - this.mode = Mode.INITIAL; + private MetricsInfoOperator( + Mode mode, + BlockFactory blockFactory, + MetricFieldLookup fieldLookup, + int metadataSourceChannel, + int indexChannel, + int[] channels + ) { + this.mode = mode; this.blockFactory = blockFactory; this.breaker = blockFactory.breaker(); this.fieldLookup = fieldLookup; this.metadataSourceChannel = metadataSourceChannel; this.indexChannel = indexChannel; - this.finalChannels = null; - } - - /** - * Creates a FINAL-mode operator (coordinator). - * - * @param channels the 6 input channel indices mapping to - * [metric_name, data_stream, unit, metric_type, field_type, dimension_fields] - */ - private MetricsInfoOperator(BlockFactory blockFactory, int[] channels) { - this.mode = Mode.FINAL; - this.blockFactory = blockFactory; - this.breaker = blockFactory.breaker(); - this.fieldLookup = null; - this.metadataSourceChannel = -1; - this.indexChannel = -1; this.finalChannels = channels; } diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java index c8727b9cb41bf..c184271847ce6 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/MetricsInfoOperatorTests.java @@ -110,29 +110,21 @@ protected void assertStatus(@Nullable Map map, List input, assertNull(map); } - private MetricsInfoOperator createInitialOperator() { - return (MetricsInfoOperator) new MetricsInfoOperator.Factory(SIMPLE_LOOKUP, METADATA_CHANNEL, INDEX_CHANNEL).get(driverContext()); + private Operator createInitialOperator() { + return createInitialOperator(SIMPLE_LOOKUP); } - private MetricsInfoOperator createInitialOperator(MetricsInfoOperator.MetricFieldLookup lookup) { - return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(driverContext()); + private Operator createInitialOperator(MetricsInfoOperator.MetricFieldLookup lookup) { + return new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(driverContext()); } - private MetricsInfoOperator createInitialOperator( - MetricsInfoOperator.MetricFieldLookup lookup, - int metadataSourceChannel, - int indexChannel - ) { - return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, metadataSourceChannel, indexChannel).get(driverContext()); - } - - private MetricsInfoOperator createFinalOperator(int[] channels) { - return (MetricsInfoOperator) new MetricsInfoOperator.FinalFactory(channels).get(driverContext()); + private Operator createFinalOperator() { + return new MetricsInfoOperator.FinalFactory(MetricsInfoOperatorTests.FINAL_CHANNELS).get(driverContext()); } public void testSingleMetricSingleIndex() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.85, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -155,7 +147,7 @@ public void testSingleMetricSingleIndex() { public void testMultipleMetricsSameIndex() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Both cpu_usage and disk_io are metric fields Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"disk_io\": 1024, \"host\": \"server1\"}", "my-index"); op.addInput(input); @@ -174,7 +166,7 @@ public void testMultipleMetricsSameIndex() { public void testSameMetricDifferentIndicesMergedBySignature() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Same metric from two different indices but with the same signature → merged into one row Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "index-a"); op.addInput(input1); @@ -202,8 +194,19 @@ public void testSameMetricDifferentIndicesMergedBySignature() { public void testDifferentSignaturesNotMerged() { DriverContext ctx = driverContext(); BlockFactory blockFactory = ctx.blockFactory(); - // Lookup returns different types for the same metric name in different indices - try (MetricsInfoOperator op = getMetricsInfoOperator(ctx)) { + + MetricsInfoOperator.MetricFieldLookup lookup = (indexName, fieldName) -> { + if ("cpu".equals(fieldName)) { + if ("index-a".equals(indexName)) { + return new MetricFieldInfo("cpu", indexName, "double", "gauge", "percent"); + } else { + return new MetricFieldInfo("cpu", indexName, "float", "gauge", "percent"); + } + } + return null; + }; + + try (Operator op = createInitialOperator(lookup)) { Page input1 = buildPage(blockFactory, "{\"cpu\": 0.5}", "index-a"); op.addInput(input1); @@ -221,23 +224,9 @@ public void testDifferentSignaturesNotMerged() { } } - private static MetricsInfoOperator getMetricsInfoOperator(DriverContext ctx) { - MetricsInfoOperator.MetricFieldLookup lookup = (indexName, fieldName) -> { - if ("cpu".equals(fieldName)) { - if ("index-a".equals(indexName)) { - return new MetricFieldInfo("cpu", indexName, "double", "gauge", "percent"); - } else { - return new MetricFieldInfo("cpu", indexName, "float", "gauge", "percent"); - } - } - return null; - }; - return (MetricsInfoOperator) new MetricsInfoOperator.Factory(lookup, METADATA_CHANNEL, INDEX_CHANNEL).get(ctx); - } - public void testDimensionFieldsCollected() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // "host" and "region" are non-metric (dimension) fields Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\", \"region\": \"us-east\"}", "my-index"); op.addInput(input); @@ -264,7 +253,7 @@ public void testNestedMetadataFields() { return null; }; - try (MetricsInfoOperator op = createInitialOperator(lookup)) { + try (Operator op = createInitialOperator(lookup)) { Page input = buildPage(blockFactory, "{\"system\": {\"cpu\": 0.75}, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -288,7 +277,7 @@ public void testNestedMetadataFields() { */ public void testDimensionFieldsUnderPassthroughStylePath() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Metadata shaped like a passthrough: resource.attributes.service.name (dimension), cpu_usage (metric) String metadataJson = """ {"resource": {"attributes": {"service": {"name": "api"}}}, "cpu_usage": 0.5} @@ -316,7 +305,7 @@ public void testDimensionFieldsUnderPassthroughStylePath() { */ public void testBothTopLevelAndPassthroughDimensionPaths() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Two documents: one with top-level service.name, one with resource.attributes.service.name Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"service\": {\"name\": \"web\"}}", "my-index"); Page input2 = buildPage( @@ -341,7 +330,7 @@ public void testBothTopLevelAndPassthroughDimensionPaths() { public void testNullMetadataSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -371,7 +360,7 @@ public void testNullMetadataSkipped() { public void testNullIndexSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -399,8 +388,7 @@ public void testNullIndexSkipped() { } public void testEmptyInputProducesEmptyPage() { - BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { op.finish(); Page output = op.getOutput(); @@ -413,8 +401,7 @@ public void testEmptyInputProducesEmptyPage() { } public void testGetOutputBeforeFinishReturnsNull() { - BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { assertNull(op.getOutput()); assertTrue(op.needsInput()); } @@ -422,7 +409,7 @@ public void testGetOutputBeforeFinishReturnsNull() { public void testGetOutputCalledTwiceReturnsNullSecondTime() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { Page input = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); op.addInput(input); op.finish(); @@ -438,8 +425,7 @@ public void testGetOutputCalledTwiceReturnsNullSecondTime() { } public void testNeedsInputReturnsFalseAfterFinish() { - BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { assertTrue(op.needsInput()); op.finish(); assertFalse(op.needsInput()); @@ -447,7 +433,7 @@ public void testNeedsInputReturnsFalseAfterFinish() { } public void testIsFinishedOnlyAfterOutputProduced() { - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { assertFalse(op.isFinished()); op.finish(); @@ -471,7 +457,7 @@ public void testMetricWithNullUnit() { return null; }; - try (MetricsInfoOperator op = createInitialOperator(lookup)) { + try (Operator op = createInitialOperator(lookup)) { Page input = buildPage(blockFactory, "{\"temperature\": 36.6}", "my-index"); op.addInput(input); op.finish(); @@ -491,7 +477,7 @@ public void testMetricWithNullUnit() { public void testInvalidJsonSkipped() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { try ( BytesRefBlock.Builder metadataBuilder = blockFactory.newBytesRefBlockBuilder(2); BytesRefBlock.Builder indexBuilder = blockFactory.newBytesRefBlockBuilder(2) @@ -521,7 +507,7 @@ public void testInvalidJsonSkipped() { public void testMultipleInputPages() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Page 1: cpu_usage from index-a Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "index-a"); op.addInput(input1); @@ -544,7 +530,7 @@ public void testMultipleInputPages() { public void testDimensionFieldsAreUnionAcrossDocuments() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Document 1: dimension "host" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); op.addInput(input1); @@ -573,8 +559,7 @@ public void testFactoryDescribe() { } public void testToString() { - BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(SIMPLE_LOOKUP, 0, 1)) { + try (Operator op = createInitialOperator()) { assertThat(op.toString(), equalTo("MetricsInfoOperator[mode=INITIAL]")); } } @@ -601,7 +586,7 @@ public void testResolveDataStreamName() { public void testDifferentMetricsHaveDifferentDimensions() { BlockFactory blockFactory = driverContext().blockFactory(); // Lookup: cpu_usage and disk_io are metrics - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Tsid 1: has cpu_usage + dimension "host" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"server1\"}", "my-index"); // Tsid 2: has disk_io + dimension "mount" @@ -639,7 +624,7 @@ public void testDifferentMetricsHaveDifferentDimensions() { */ public void testBackingIndicesSameDataStreamGroupedTogether() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Two backing indices of the same data stream "k8s" Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu_usage\": 0.9, \"host\": \"b\"}", ".ds-k8s-2024.01.15-000002"); @@ -759,7 +744,7 @@ private static MetricsInfoOperator getOperator(DriverContext ctx) { */ public void testDifferentDataStreamsSameValuesProduceMultiValuedDataStream() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator()) { + try (Operator op = createInitialOperator()) { // Two different data streams, same metric with the same signature Page input1 = buildPage(blockFactory, "{\"cpu_usage\": 0.5, \"host\": \"a\"}", ".ds-k8s-2024.01.15-000001"); Page input2 = buildPage(blockFactory, "{\"cpu_usage\": 0.9, \"host\": \"b\"}", ".ds-other-2024.01.15-000001"); @@ -782,7 +767,7 @@ public void testDifferentDataStreamsSameValuesProduceMultiValuedDataStream() { public void testFinalModeIdenticalRowsFromTwoDataNodesAreMerged() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { // Two data nodes produced identical rows for cpu_usage on index-a Page page1 = buildFinalPage( blockFactory, @@ -820,7 +805,7 @@ public void testFinalModeIdenticalRowsFromTwoDataNodesAreMerged() { public void testFinalModePartiallyOverlappingDataStreamsAreMerged() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { // Data node 1 saw index-a and index-b; data node 2 saw index-a and index-c Page page1 = buildFinalPage( blockFactory, @@ -856,7 +841,7 @@ public void testFinalModePartiallyOverlappingDataStreamsAreMerged() { public void testFinalModeDifferentMetricNamesRemainSeparateRows() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { Page page1 = buildFinalPage( blockFactory, "cpu_usage", @@ -891,7 +876,7 @@ public void testFinalModeDifferentMetricNamesRemainSeparateRows() { public void testFinalModeDifferentSignaturesProduceSeparateRows() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { // Same metric name but different field_type → different signatures → 2 rows Page page1 = buildFinalPage( blockFactory, @@ -925,8 +910,7 @@ public void testFinalModeDifferentSignaturesProduceSeparateRows() { } public void testFinalModeEmptyInputProducesEmptyOutput() { - BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { op.finish(); Page output = op.getOutput(); @@ -940,7 +924,7 @@ public void testFinalModeEmptyInputProducesEmptyOutput() { public void testFinalModeSingleDataNodeInputPassesThrough() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { Page page = buildFinalPage( blockFactory, "cpu_usage", @@ -970,7 +954,7 @@ public void testFinalModeSingleDataNodeInputPassesThrough() { public void testFinalModeDimensionFieldsAreUnionedAcrossDataNodes() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { // Data node 1 saw dimension "host"; data node 2 saw dimension "region" Page page1 = buildFinalPage( blockFactory, @@ -1012,7 +996,7 @@ public void testFinalModeDimensionFieldsAreUnionedAcrossDataNodes() { */ public void testFinalModeSameDataStreamDifferentUnitsAcrossNodesProduceOneRow() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createFinalOperator(FINAL_CHANNELS)) { + try (Operator op = createFinalOperator()) { // Data node A has backing index 000001 (unit=usd) Page page1 = buildFinalPage( blockFactory, @@ -1114,7 +1098,7 @@ private static void appendSet(BytesRefBlock.Builder builder, Set values) */ public void testHistogramMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { + try (Operator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"legacy": {"values": [1.0, 2.0, 3.0], "counts": [10, 20, 30]}}, "entity": {"id": "abc"}}"""; Page input = buildPage(blockFactory, metadataJson.trim(), ".ds-histograms-2026.02.16-000001"); @@ -1144,7 +1128,7 @@ public void testHistogramMetricRecognizedFromNestedSyntheticSource() { */ public void testExponentialHistogramMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { + try (Operator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"exponential": {"scale": 3, "sum": 42.5, "min": 1.0, "max": 10.0,\ "zero": {"count": 0, "threshold": 0.0},\ @@ -1177,7 +1161,7 @@ public void testExponentialHistogramMetricRecognizedFromNestedSyntheticSource() */ public void testTdigestMetricRecognizedFromNestedSyntheticSource() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { + try (Operator op = createInitialOperator(HISTOGRAM_LOOKUP)) { String metadataJson = """ {"histogram": {"tdigest": {"min": 0.5, "max": 99.5, "sum": 500.0,\ "centroids": [1.0, 50.0, 99.0], "counts": [100, 200, 100]}},\ @@ -1208,7 +1192,7 @@ public void testTdigestMetricRecognizedFromNestedSyntheticSource() { */ public void testMixedScalarAndHistogramMetrics() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { + try (Operator op = createInitialOperator(HISTOGRAM_LOOKUP)) { // cpu_usage is a scalar metric (leaf), histogram.legacy is a complex metric (nested Map) String metadataJson = """ {"cpu_usage": 0.75, "histogram": {"legacy": {"values": [1.0], "counts": [5]}}, "host": "srv1"}"""; @@ -1241,7 +1225,7 @@ public void testMixedScalarAndHistogramMetrics() { */ public void testDimensionsNotContaminatedByHistogramInternalFields() { BlockFactory blockFactory = driverContext().blockFactory(); - try (MetricsInfoOperator op = createInitialOperator(HISTOGRAM_LOOKUP)) { + try (Operator op = createInitialOperator(HISTOGRAM_LOOKUP)) { // All three histogram types in one document, plus a dimension String metadataJson = """ {"histogram": {\