This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 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 396e7211280 Add per-key segment build observability for OPEN_STRUCT
columns (PR 4/4) (#19041)
396e7211280 is described below
commit 396e72112801576412cb73d264b727e66b484ef3
Author: tarun11Mavani <[email protected]>
AuthorDate: Tue Aug 25 02:35:24 2026 +0530
Add per-key segment build observability for OPEN_STRUCT columns (PR 4/4)
(#19041)
* feat(open_struct): per-key segment build observability metrics
Add per-key segment build observability for OPEN_STRUCT columns.
Metric cardinality is bounded to configured dense keys (not data-driven
key discovery) to prevent unbounded registry growth.
Meters (column-level):
- OPEN_STRUCT_TYPE_COERCION_FAILURES
- OPEN_STRUCT_TYPE_INFERENCE_FAILURES
Gauges (column-level):
- OPEN_STRUCT_DENSE_KEY_COUNT
- OPEN_STRUCT_SPARSE_KEY_COUNT
- OPEN_STRUCT_TOTAL_KEYS_DISCOVERED
Gauge (per dense key):
- OPEN_STRUCT_KEY_FILL_RATE
* fix(open_struct): address review feedback on build observability metrics
Reworks the OPEN_STRUCT segment-build metrics from PR review.
Scrape rules (server.yml):
Adds three rules ahead of the generic ones. The per-key gauge previously
matched no rule at all; the column-level gauges and meters did match, but
the generic "tableNameWithType + partitionId" and "rawTableName" rules
claimed them first and exported the column as partition="<column>" and
table="<table>.<column>" respectively. Keys are captured raw rather than
sanitised: '$', '.' and spaces are all legal in a Prometheus label value,
and sanitising would collapse user.id, user-id and user_id.
Fill rate:
Replaces OPEN_STRUCT_KEY_FILL_RATE with OPEN_STRUCT_KEY_DOC_COUNT and
OPEN_STRUCT_SEGMENT_DOC_COUNT. Integer division truncated a dense key
present in a handful of docs to 0, indistinguishable from no data and
exactly the case worth alerting on. No alias is kept: the gauge was never
released upstream. ServerGauge.DOCUMENT_COUNT is unsuitable as the
denominator (accumulating server-wide total, keyed on table only, emitted
at segment load rather than build).
Type-inference fallback:
Extracts OpenStructTypeInference#resolve, shared by the consuming and the
sealed build paths so the fallback rule cannot drift. A value that maps to
no DataType is stored as its serialized string form on both paths; dropping
on one made a doc resolve differently either side of the seal boundary and
made the REALTIME and OFFLINE halves of a hybrid table disagree.
Failure counting:
Counts per value rather than once per key, since the key's inferred type is
cached on first sighting. A value that is unmappable but lands on a key
already typed non-STRING is dropped by coercion and counted only there --
counting it in both meters recorded one dropped value twice and made the
seal-time log claim it fell back to STRING when it had not. Meters are
keyed on the column, not the key: they fire on malformed input, so an
id-like key would mint one meter per id and meters are never removed from
the registry.
Logging:
Seal-time summaries log a total plus the top 5 offending keys at INFO and
the full map at DEBUG; the key space is user-controlled. The consuming-path
WARN fires only on a key's first sighting.
Tests:
Covers per-value counting, the unmappable-on-typed-key case, gauge emission
including a fill rate that would truncate to zero, consuming/sealed parity
for Map and List values, top-N capping, and the scrape rules -- including a
first-match-wins test over the whole ordered rule list, since the ordering
is what makes the new rules correct. PrometheusTemplateRegexpTest selected
rules by name via findFirst, but rule names are only the exported
metric-name template and are not unique; it now requires a unique match and
takes a discriminator.
* test(open_struct): cover the exported shape of the OPEN_STRUCT metrics
PrometheusTemplateRegexpTest only runs Pattern.matches() against strings; it
never exercises the name:/labels: substitution. ServerPrometheusMetricsTest
is
the one that starts a real JmxCollector over server.yml and scrapes, and
being
data-provider-driven over ServerGauge.values() it already ran the new
gauges --
but through the default branch, which emits a plain tableNameWithType. That
produces openStructKeyDocCount.<table>_REALTIME with no .column$key
segment, a
name the OPEN_STRUCT rule cannot match, so a generic rule claimed it and the
assertion passed against a shape that never occurs in production. The two
column-level meters had the same problem: no column segment, so their rule
never fired either.
Dispatches all seven separately, emitting through the keyed APIs with the
real
third name segment built via OpenStructNaming#materializedColumnName, and
asserts the full label set including column and key.
The test key is "clicks.v2$promo-code". The embedded '$' is the case worth
pinning, since '$' is the column/key delimiter and only a column group that
stops at the first one plus a greedy key group can round-trip it. Spaces are
equally legal in a label value but are not covered:
PromMetric#fromExportedMetric
splits the scrape line on the first space, so that defeats the harness
rather
than the exporter.
Verified by deleting the OPEN_STRUCT rule block from server.yml, which fails
exactly these seven and nothing else. Runs under Yammer only -- the
Dropwizard
suite is disabled repo-wide.
* fix(open_struct): bound the per-key gauge to configured dense keys
Addresses the two open points from review.
Per-key gauge cardinality:
OPEN_STRUCT_KEY_DOC_COUNT now iterates the configured denseKeys rather
than _resolvedDenseKeys. The earlier "config-bounded" claim was wrong:
DEFAULT_MAX_DENSE_KEYS is -1 and denseKeys defaults to empty, so under the
default config _resolvedDenseKeys is every key at or above the 0.5 fill
threshold -- data-driven, and gauges are never removed from the registry.
maxDenseKeys > 0 is not a sufficient gate either, since it caps each
segment's set but different segments resolve different keys, leaving the
union across segments unbounded. An explicit denseKeys list is the only
operator-owned bound.
A configured key absent from the segment is skipped rather than reported
as 0. One that appeared but lost its slot to the maxDenseKeys cap still
reports: a configured key not earning its materialized column is the case
worth seeing. materializedColumnName stays the source of the metric key in
that case -- it is only the "<column>$<key>" identifier the scrape rule
splits back into column and key labels, not a claim of an on-disk column,
and a second helper with the same format would drift from parseParentColumn
and the server.yml rule.
Gauge naming:
Renames OPEN_STRUCT_TOTAL_KEYS_DISCOVERED to OPEN_STRUCT_SEGMENT_KEY_COUNT.
"Total ... discovered" read as a table-wide cumulative count, but all five
gauges are keyed on (table, column) with no segment dimension, so each seal
overwrites the last and they mean "the most recently sealed segment". The
new name matches the OPEN_STRUCT_SEGMENT_DOC_COUNT convention already
established for that scope. No alias: the gauge never reached master.
Tests:
Four cases over the new gating -- nothing emitted under default config,
only configured keys emitted when others also resolve dense from data, a
configured key cut by the maxDenseKeys cap still emitted, and a configured
key absent from the segment skipped.
* fix(open_struct): escape JMX-reserved chars in per-key metric names
OPEN_STRUCT keys come from user JSON. ObjectName.quote backslash-escapes
exactly '"', '\', '*' and '?' on the way to JMX. An escaped '"' stops the
exported name matching the per-key rule in server.yml at all, so the metric
silently loses its column/key labels and falls through to a generic rule;
the other three match but leave a stray backslash in the label value.
Add OpenStructNaming#metricKey, which percent-escapes those four (plus '%'
itself, so the mapping stays injective) and is used at the per-key gauge
emission site. Folding them onto '_' would collide: '_' is a legal key
character, so 'a"b' and 'a_b' would share one gauge and overwrite each
other on every seal.
Escaping is deliberately narrow -- '.', '-', '_', ',', '=' and spaces are
safe in a quoted ObjectName and in a Prometheus label value, so 'user.id',
'user-id' and 'user_id' stay distinct series.
Tests: ServerPrometheusMetricsTest now drives a '"'-bearing key end to end
through the JMX exporter; OpenStructColumnSplitterTest pins the emission
site to metricKey. Both fail if the escaping is reverted.
* refactor(open_struct): trim the observability diff
No behaviour change.
- OpenStructNaming: replace the 27-line hand-rolled escape loop with chained
String.replace, '%' first so the escapes it introduces are not re-escaped.
Verified byte-identical to the loop over 400k random keys drawn from
{%, ", \, *, ?, a, ., -, _, $}.
- OpenStructColumnSplitter: drop topFailures and MAX_LOGGED_FAILURE_KEYS.
The
INFO line already carries the total and the key count, and the full
per-key
map is logged at DEBUG; the top-5 tier in between was not worth its own
method, constant and test. Also drop a local alias for _columnName.
- AbstractMetrics: compress the removeKeyedTableGauges comment.
* docs(open_struct): trim javadoc on the new observability symbols
Doc-only; no code lines changed.
Cuts rationale that belongs in the PR discussion rather than the API
contract: the paragraph defending percent-escaping over folding to '_',
the note on revisiting the escape set if metrics stop going through JMX,
the "plain class rather than a record" aside, and the restated reasons
keyed gauge names cannot be recovered from the registry.
* fix(open_struct): remove per-key gauges when the table is deleted
OPEN_STRUCT_KEY_DOC_COUNT registers as <gauge>.<table>.<column>$<key>. The
sweep in TableDeletionMessageHandler composes only the unkeyed
<gauge>.<table>, so it could never reach these and they outlived the table.
Re-derive the key set from the table config and remove each one explicitly.
This works because the gauge is emitted only for explicitly configured
denseKeys rather than for the ingested key space, so the set is recoverable
from the config. Keys are resolved before deleteTable() runs, since the only
in-memory copy of the config lives on the table data manager that call is
about to discard; resolution failures are logged rather than allowed to
block deletion.
Keys are derived from the current config, so a key dropped from denseKeys
after a segment already emitted its gauge is not covered -- that gauge is
orphaned by the config change itself. Closing that gap needs the emitted
names tracked at emit time, which is left for a follow-up.
* refactor(open_struct): qualify the per-segment gauges with LAST_SEGMENT
These gauges are written with setOrUpdateTableGauge at segment build time,
so each build overwrites the previous value for the same (table, column).
They read as "the most recently built segment", not as a table-wide total,
but the names did not say so -- unlike the LAST_REALTIME_SEGMENT_* gauges
alongside them. A dashboard summing openStructDenseKeyCount across replicas
would have gotten a silently meaningless number. Enum names are permanent
once shipped, so this lands before the metrics do.
openStructDenseKeyCount -> openStructLastSegmentDenseKeyCount
openStructSparseKeyCount -> openStructLastSegmentSparseKeyCount
openStructSegmentKeyCount -> openStructLastSegmentKeyCount
openStructSegmentDocCount -> openStructLastSegmentDocCount
openStructKeyDocCount -> openStructLastSegmentKeyDocCount
Pure rename plus the matching scrape rules in server.yml; the block comment
explaining the semantic shrinks now that the names carry it.
* feat(open_struct): emit the per-key doc-count gauge for every ingested key
OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT fired only for keys named in the
table's denseKeys config. That answers "is a key I already declared
earning its materialized column", but not "which key should I declare",
which is the question the fill rate is actually useful for. Emit for
every key present in the sealed segment instead -- dense or sparse,
configured or discovered.
The cost is that the registry entry count now follows the ingested key
space rather than the config, and table deletion can only sweep the keys
recoverable from denseKeys, so gauges for discovered keys survive until
the server restarts. Gating this emission behind a config switch is the
follow-up; tracking emitted key names at emit time is the alternative to
re-deriving them from config at deletion.
* feat(open_struct): gate per-key doc-count gauge behind
perKeyDocCountEnabled (default off)
* fix(open_struct): make per-key type resolution sticky once non-STRING
Once a key's inferred type is established as non-STRING, a later value
that fails OpenStructTypeInference forced valueType back to STRING for
that single value, desyncing it from _inferredTypes and corrupting
_values with a mixed-type list for the key instead of being dropped as
a coercion failure. Skip inference entirely once established is
non-STRING, matching MutableOpenStructIndex's fast path; the value
falls through to the existing coercion try/catch instead.
testUnmappableValueOnTypedKeyCountsOnlyAsCoercionFailure already pins
this behavior and fails with a ClassCastException without this fix.
* refactor(open_struct): drop dead condition in inference-failure counting
The sticky non-STRING branch above already guarantees `established` is
null or STRING by the time this else is reached, so the re-check was
always true. No behavior change.
* refactor(open_struct): rename perKeyDocCountEnabled to
perKeyMetricsEnabled
The flag now gates more than the doc-count gauge, so name it after what
it controls at the config level rather than the first metric it enabled.
* chore: retrigger CI (infra flakiness on codeload.github.com)
---
.../jmx_prometheus_javaagent/configs/server.yml | 34 +++
.../apache/pinot/common/metrics/ServerGauge.java | 20 +-
.../apache/pinot/common/metrics/ServerMeter.java | 7 +-
.../prometheus/PrometheusTemplateRegexpTest.java | 158 ++++++++++-
.../prometheus/ServerPrometheusMetricsTest.java | 49 ++++
.../indexsegment/mutable/MutableSegmentImpl.java | 3 +-
.../impl/openstruct/OpenStructColumnSplitter.java | 113 ++++++--
.../index/openstruct/MutableOpenStructIndex.java | 60 +++--
.../index/openstruct/OpenStructIndexType.java | 3 +-
.../openstruct/OpenStructColumnSplitterTest.java | 294 +++++++++++++++++++--
.../MutableOpenStructDataSourceTest.java | 34 +--
.../openstruct/MutableOpenStructIndexTest.java | 72 ++++-
.../OpenStructConsumingSealedParityTest.java | 78 +++++-
.../helix/SegmentMessageHandlerFactory.java | 54 ++++
.../helix/SegmentMessageHandlerFactoryTest.java | 188 +++++++++++++
.../spi/config/table/OpenStructIndexConfig.java | 24 +-
.../apache/pinot/spi/data/OpenStructNaming.java | 20 ++
.../config/table/OpenStructIndexConfigTest.java | 22 ++
.../pinot/spi/data/OpenStructNamingTest.java | 64 +++++
19 files changed, 1197 insertions(+), 100 deletions(-)
diff --git
a/docker/images/pinot/etc/jmx_prometheus_javaagent/configs/server.yml
b/docker/images/pinot/etc/jmx_prometheus_javaagent/configs/server.yml
index 4d029798bd2..05c829e7642 100644
--- a/docker/images/pinot/etc/jmx_prometheus_javaagent/configs/server.yml
+++ b/docker/images/pinot/etc/jmx_prometheus_javaagent/configs/server.yml
@@ -1,4 +1,38 @@
rules:
+# OPEN_STRUCT observability. These MUST precede the generic gauge/meter rules
below: the column segment
+# would otherwise be consumed by the "tableNameWithType + partitionId" rule
and exported as partition="<column>",
+# and the column-level meters by the "rawTableName" rule as
table="<table>.<column>".
+# Per-key gauge. The key is a raw user-supplied JSON key, so it is captured
last with a greedy ([^"]+) — it may
+# contain '$', '.', '-' or spaces, all of which are legal in a Prometheus
label value. The closing quote is
+# excluded because the trailing "? is optional and a greedy (.+) would
otherwise swallow it. The column name
+# cannot contain '.' or '$', so ([^.$"]+) splits the two unambiguously. The
sparse column renders as key="__sparse__".
+- pattern:
"\"?org\\.apache\\.pinot\\.common\\.metrics\"?<type=\"ServerMetrics\",
name=\"?pinot\\.server\\.(openStructLastSegmentKeyDocCount)\\.(([^.]+)\\.)?([^.]*)_(OFFLINE|REALTIME)\\.([^.$\"]+)\\$([^\"]+)\"?><>(\\w+)"
+ name: "pinot_server_$1_$8"
+ cache: true
+ labels:
+ database: "$3"
+ table: "$2$4"
+ tableType: "$5"
+ column: "$6"
+ key: "$7"
+# Column-level OPEN_STRUCT gauges.
+- pattern:
"\"?org\\.apache\\.pinot\\.common\\.metrics\"?<type=\"ServerMetrics\",
name=\"?pinot\\.server\\.(openStructLastSegmentDenseKeyCount|openStructLastSegmentSparseKeyCount|openStructLastSegmentKeyCount|openStructLastSegmentDocCount)\\.(([^.]+)\\.)?([^.]*)_(OFFLINE|REALTIME)\\.([^.\"]+)\"?><>(\\w+)"
+ name: "pinot_server_$1_$7"
+ cache: true
+ labels:
+ database: "$3"
+ table: "$2$4"
+ tableType: "$5"
+ column: "$6"
+# Column-level OPEN_STRUCT meters. Meter names put the metric name last,
unlike gauges.
+- pattern:
"\"?org\\.apache\\.pinot\\.common\\.metrics\"?<type=\"ServerMetrics\",
name=\"?pinot\\.server\\.(([^.]+)\\.)?([^.]*)_(OFFLINE|REALTIME)\\.([^.]+)\\.(openStructTypeCoercionFailures|openStructTypeInferenceFailures)\"?><>(\\w+)"
+ name: "pinot_server_$6_$7"
+ cache: true
+ labels:
+ database: "$2"
+ table: "$1$3"
+ tableType: "$4"
+ column: "$5"
# Gauges that accept tableNameWithType
- pattern: "\"?org\\.apache\\.pinot\\.common\\.metrics\"?<type=\"?\\w+\"?,
name=\"?pinot\\.(\\w+)\\.(\\w+)\\.((\\w+)\\.)?(\\w+)_(OFFLINE|REALTIME)\\\"?><>(\\w+)"
name: "pinot_$1_$2_$7"
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
index 75b68684beb..878693969dd 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
@@ -168,7 +168,25 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
THROTTLE_EXECUTOR_QUEUE_SIZE("count", true,
"Current number of tasks in the throttle executor queue"),
// Workload config fetch status: 1 = success, 0 = failure
- WORKLOAD_CONFIG_FETCH_STATUS("status", true);
+ WORKLOAD_CONFIG_FETCH_STATUS("status", true),
+ // OPEN_STRUCT segment build observability. Each seal on this server
overwrites the previous value for the
+ // same (table, column), so these are not table-wide totals and must not be
summed across segments or
+ // replicas; sampling them over time gives the trend.
+ OPEN_STRUCT_LAST_SEGMENT_DENSE_KEY_COUNT("keys", false,
+ "Number of OPEN_STRUCT keys classified as dense in the most recently
sealed segment"),
+ OPEN_STRUCT_LAST_SEGMENT_SPARSE_KEY_COUNT("keys", false,
+ "Number of OPEN_STRUCT keys classified as sparse in the most recently
sealed segment"),
+ OPEN_STRUCT_LAST_SEGMENT_KEY_COUNT("keys", false,
+ "Unique keys in the most recently sealed segment for this OPEN_STRUCT
column (dense + sparse)"),
+ OPEN_STRUCT_LAST_SEGMENT_DOC_COUNT("documents", false,
+ "Total docs in the most recently sealed segment for this OPEN_STRUCT
column; denominator for "
+ + "openStructLastSegmentKeyDocCount"),
+ /// When `perKeyMetricsEnabled` is false (default), emitted only for keys
named in `denseKeys`. When
+ /// true, emitted for every key in the sealed segment — registry entries
follow the ingested key space.
+ /// See `OpenStructColumnSplitter#emitMetrics`.
+ OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT("documents", false,
+ "Docs in which an OPEN_STRUCT key was present in the most recently
sealed segment; "
+ + "divide by openStructLastSegmentDocCount for the fill rate");
private final String _gaugeName;
private final String _unit;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
index 9020607da71..055d2cb2cbe 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
@@ -240,7 +240,12 @@ public enum ServerMeter implements AbstractMetrics.Meter {
DROPPED_RECORD_COUNT("rows", false),
CORRUPTED_RECORD_COUNT("rows", false),
OPEN_STRUCT_TYPE_COERCION_FAILURES("values", false,
- "Number of OPEN_STRUCT values dropped because the value could not be
coerced to the key's inferred type"),
+ "Number of OPEN_STRUCT values dropped because the value could not be
coerced to the key's declared or "
+ + "already-established type"),
+ OPEN_STRUCT_TYPE_INFERENCE_FAILURES("values", false,
+ "Number of OPEN_STRUCT values stored as their serialized string form
because the value's Java type maps to "
+ + "no Pinot DataType. A value that is instead dropped for failing
coercion against a key's "
+ + "already-established non-STRING type is counted by
openStructTypeCoercionFailures, not here"),
// Workload related metrics
WORKLOAD_BUDGET_EXCEEDED("workloadBudgetExceeded", true, "Number of times
workload budget exceeded"),
WORKLOAD_QUERIES("queries", false),
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/PrometheusTemplateRegexpTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/PrometheusTemplateRegexpTest.java
index 13d69065d5d..19c665f7a05 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/PrometheusTemplateRegexpTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/PrometheusTemplateRegexpTest.java
@@ -172,7 +172,7 @@ public class PrometheusTemplateRegexpTest {
@Test
public void testServerTableWithTypeAndPartitionGaugePattern()
throws Exception {
- String pattern = loadPatternByName("server.yml", "pinot_server_$1_$7");
+ String pattern = loadPatternByName("server.yml", "pinot_server_$1_$7",
"pinot\\.server\\.(\\w+)\\.");
Matcher m = Pattern.compile(pattern).matcher(
"\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ "name=\"pinot.server.queries.myTable_REALTIME.3\"><>Value");
@@ -184,6 +184,134 @@ public class PrometheusTemplateRegexpTest {
Assert.assertEquals(m.group(7), "Value");
}
+ // ---- OPEN_STRUCT server patterns ----
+
+ /// server.yml: per-key OPEN_STRUCT gauge. The JMX name embeds the raw
user-supplied JSON key
+ /// after the `$` separator, so the key group must survive characters the
generic `\w+` rules
+ /// reject: `$`, `.`, `-` and spaces are all legal in a Prometheus label
value.
+ @Test
+ public void testServerOpenStructPerKeyGaugePattern()
+ throws Exception {
+ Pattern compiled = Pattern.compile(
+ loadPatternByName("server.yml", "pinot_server_$1_$8",
"openStructLastSegmentKeyDocCount"));
+
+ Matcher plain =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentKeyDocCount.myTable_REALTIME.metrics$clicks\"><>Value");
+ Assert.assertTrue(plain.matches(), "Pattern should match per-key
OPEN_STRUCT gauge");
+ Assert.assertEquals(plain.group(1), "openStructLastSegmentKeyDocCount");
+ Assert.assertEquals(plain.group(4), "myTable");
+ Assert.assertEquals(plain.group(5), "REALTIME");
+ Assert.assertEquals(plain.group(6), "metrics");
+ Assert.assertEquals(plain.group(7), "clicks");
+ Assert.assertEquals(plain.group(8), "Value");
+
+ // A key containing '.' would be swallowed by the generic rules; the
column group is ([^.$]+)
+ // so the split stays unambiguous and the whole remainder lands in the key
label.
+ Matcher dotted =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentKeyDocCount.myDb.myTable_OFFLINE.metrics$user.id\"><>Value");
+ Assert.assertTrue(dotted.matches(), "Pattern should match a key containing
'.'");
+ Assert.assertEquals(dotted.group(3), "myDb");
+ Assert.assertEquals(dotted.group(4), "myTable");
+ Assert.assertEquals(dotted.group(6), "metrics");
+ Assert.assertEquals(dotted.group(7), "user.id");
+
+ // A key containing '$' — greedy (.+) puts the split at the first '$',
which is the separator
+ // the splitter emitted, so the trailing '$' stays part of the key.
+ Matcher dollar =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentKeyDocCount.myTable_OFFLINE.metrics$a$b\"><>Value");
+ Assert.assertTrue(dollar.matches(), "Pattern should match a key containing
'$'");
+ Assert.assertEquals(dollar.group(6), "metrics");
+ Assert.assertEquals(dollar.group(7), "a$b");
+
+ // The sparse catch-all column is named with the reserved __sparse__
suffix rather than a key.
+ Matcher sparse =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentKeyDocCount.myTable_OFFLINE.metrics$__sparse__\"><>Value");
+ Assert.assertTrue(sparse.matches(), "Pattern should match the sparse
column");
+ Assert.assertEquals(sparse.group(7), "__sparse__");
+ }
+
+ /// server.yml: column-level OPEN_STRUCT gauges. These must be matched here
rather than by the
+ /// generic "tableNameWithType + partitionId" rule, which would export the
column as
+ /// partition="<column>". Every branch of the alternation is exercised: a
typo in one of them is
+ /// still a valid regexp, so it would pass testAllPatternsAreValidRegexp and
then silently fail to
+ /// scrape in production.
+ @Test
+ public void testServerOpenStructColumnGaugePattern()
+ throws Exception {
+ Pattern compiled =
+ Pattern.compile(loadPatternByName("server.yml", "pinot_server_$1_$7",
"openStructLastSegmentDenseKeyCount"));
+ for (String metric : List.of("openStructLastSegmentDenseKeyCount",
"openStructLastSegmentSparseKeyCount",
+ "openStructLastSegmentKeyCount", "openStructLastSegmentDocCount")) {
+ Matcher m =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ + "name=\"pinot.server." + metric +
".myTable_OFFLINE.metrics\"><>Value");
+ Assert.assertTrue(m.matches(), "Pattern should match column-level
OPEN_STRUCT gauge " + metric);
+ Assert.assertEquals(m.group(1), metric);
+ Assert.assertEquals(m.group(4), "myTable");
+ Assert.assertEquals(m.group(5), "OFFLINE");
+ Assert.assertEquals(m.group(6), "metrics");
+ Assert.assertEquals(m.group(7), "Value");
+ }
+ }
+
+ /// server.yml: column-level OPEN_STRUCT meters. Meter JMX names put the
metric name last,
+ /// unlike gauges, so these need their own rule ahead of the generic
rawTableName meter rule
+ /// (which would otherwise export table="<table>.<column>").
+ @Test
+ public void testServerOpenStructColumnMeterPattern()
+ throws Exception {
+ Pattern compiled = Pattern.compile(
+ loadPatternByName("server.yml", "pinot_server_$6_$7",
"openStructTypeCoercionFailures"));
+ for (String metric : List.of("openStructTypeCoercionFailures",
"openStructTypeInferenceFailures")) {
+ Matcher m =
compiled.matcher("\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ + "name=\"pinot.server.myTable_REALTIME.metrics." + metric +
"\"><>Count");
+ Assert.assertTrue(m.matches(), "Pattern should match column-level
OPEN_STRUCT meter " + metric);
+ Assert.assertEquals(m.group(3), "myTable");
+ Assert.assertEquals(m.group(4), "REALTIME");
+ Assert.assertEquals(m.group(5), "metrics");
+ Assert.assertEquals(m.group(6), metric);
+ Assert.assertEquals(m.group(7), "Count");
+ }
+ }
+
+ /// jmx_exporter evaluates rules in file order and stops at the first match,
so the OPEN_STRUCT
+ /// rules are only correct because they precede the generic ones. Asserting
the pattern in
+ /// isolation does not cover that: the generic "tableNameWithType +
partitionId" gauge rule and
+ /// the generic rawTableName meter rule both full-match these names too, and
would export the
+ /// column as partition="metrics" and table="myTable.metrics" respectively.
This test evaluates
+ /// the whole ordered list so a future reordering of server.yml fails here
rather than in prod.
+ @Test
+ public void testServerOpenStructRulesPrecedeGenericRules()
+ throws Exception {
+ List<String> ordered = extractPatterns(CONFIG_BASE_PATH + "/server.yml");
+ assertFirstMatchingPatternContains(ordered,
+ "\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentKeyDocCount.myTable_OFFLINE.metrics$clicks\"><>Value",
+ "openStructLastSegmentKeyDocCount");
+ assertFirstMatchingPatternContains(ordered,
+ "\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.openStructLastSegmentDenseKeyCount.myTable_OFFLINE.metrics\"><>Value",
+ "openStructLastSegmentDenseKeyCount");
+ assertFirstMatchingPatternContains(ordered,
+ "\"org.apache.pinot.common.metrics\"<type=\"ServerMetrics\", "
+ +
"name=\"pinot.server.myTable_REALTIME.metrics.openStructTypeInferenceFailures\"><>Count",
+ "openStructTypeInferenceFailures");
+ }
+
+ /// Asserts the first rule in file order that matches `jmxName` is one whose
pattern contains
+ /// `expectedInPattern`, mirroring jmx_exporter's first-match-wins
evaluation.
+ private void assertFirstMatchingPatternContains(List<String>
orderedPatterns, String jmxName,
+ String expectedInPattern) {
+ for (String pattern : orderedPatterns) {
+ if (Pattern.compile(pattern).matcher(jmxName).matches()) {
+ Assert.assertTrue(pattern.contains(expectedInPattern),
+ "First rule matching [" + jmxName + "] was [" + pattern + "],
expected one containing '"
+ + expectedInPattern + "'. OPEN_STRUCT rules must precede the
generic rules in server.yml.");
+ return;
+ }
+ }
+ Assert.fail("No rule in server.yml matches [" + jmxName + "]");
+ }
+
// ---- Controller patterns ----
/// controller.yml: minion task-type gauge.
@@ -252,23 +380,39 @@ public class PrometheusTemplateRegexpTest {
Assert.assertEquals(m.group(3), "Value");
}
- /// Returns the pattern string for the rule whose `name` field equals
`ruleName`.
+ /// Returns the pattern string for the rule whose `name` field equals
`ruleName`. Fails when more
+ /// than one rule shares that name — use
[#loadPatternByName(String,String,String)] instead.
+ ///
/// Keying off the rule name survives YAML rule reorderings — inserting or
moving a rule in
/// the config file will not silently shift the index and cause this test to
assert against
/// the wrong pattern.
- @SuppressWarnings("unchecked")
private String loadPatternByName(String configFile, String ruleName)
throws Exception {
+ return loadPatternByName(configFile, ruleName, "");
+ }
+
+ /// Same as [#loadPatternByName(String,String)], but narrowed to the rule
whose pattern also
+ /// contains `patternDiscriminator`. Rule names are only the exported
metric-name template
+ /// (e.g. `pinot_server_$1_$7`), so several rules can legitimately share
one; the discriminator
+ /// picks the intended rule instead of silently taking whichever comes first
in the file.
+ @SuppressWarnings("unchecked")
+ private String loadPatternByName(String configFile, String ruleName, String
patternDiscriminator)
+ throws Exception {
Yaml yaml = new Yaml();
try (FileReader reader = new FileReader(CONFIG_BASE_PATH + "/" +
configFile)) {
Map<String, Object> config = yaml.load(reader);
List<Map<String, Object>> rules = (List<Map<String, Object>>)
config.get("rules");
- return rules.stream()
+ List<String> matches = rules.stream()
.filter(rule -> ruleName.equals(rule.get("name")))
.map(rule -> (String) rule.get("pattern"))
- .findFirst()
- .orElseThrow(() -> new IllegalArgumentException(
- "No rule with name '" + ruleName + "' found in " + configFile));
+ .filter(pattern -> pattern != null &&
pattern.contains(patternDiscriminator))
+ .collect(Collectors.toList());
+ Assert.assertFalse(matches.isEmpty(),
+ "No rule named '" + ruleName + "' containing '" +
patternDiscriminator + "' in " + configFile);
+ Assert.assertEquals(matches.size(), 1,
+ "Ambiguous rule name '" + ruleName + "' in " + configFile
+ + "; pass a patternDiscriminator to select one of: " + matches);
+ return matches.get(0);
}
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/ServerPrometheusMetricsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/ServerPrometheusMetricsTest.java
index 274a415c55a..fc556d84f20 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/ServerPrometheusMetricsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/metrics/prometheus/ServerPrometheusMetricsTest.java
@@ -28,6 +28,7 @@ import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.metrics.ServerTimer;
+import org.apache.pinot.spi.data.OpenStructNaming;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -66,6 +67,43 @@ public abstract class ServerPrometheusMetricsTest extends
PinotPrometheusMetrics
List.of(ServerGauge.REALTIME_OFFHEAP_MEMORY_USED,
ServerGauge.REALTIME_SEGMENT_NUM_PARTITIONS,
ServerGauge.LUCENE_INDEXING_DELAY_MS,
ServerGauge.LUCENE_INDEXING_DELAY_DOCS);
+ // OPEN_STRUCT metrics carry a third name segment that the generic rules
cannot parse: "<column>" for the
+ // column-level ones and "<column>$<key>" for the per-key gauge. Emitting
them with a plain
+ // tableNameWithType (what the default branches below do) produces a name
the OPEN_STRUCT rules do not
+ // match at all, so a generic rule claims it and the assertion passes
against a shape that never occurs in
+ // production. They are dispatched separately so the exported labels are
checked against the real name.
+ private static final String OPEN_STRUCT_COLUMN = "metrics";
+ // The key as ingested. Exercises an embedded '$', a '.', a '-' and a '"'.
The '$' is worth pinning
+ // because it is the column/key delimiter, so only a column group that stops
at the first '$' and a
+ // greedy key group that takes the rest can round-trip it. The '"' is worth
pinning because
+ // ObjectName.quote backslash-escapes it on the way to JMX, which stops the
name matching the scrape
+ // rule at all — the metric would silently lose its column/key labels.
OpenStructNaming#metricKey
+ // percent-escapes it before emission, so the exported label is
OPEN_STRUCT_KEY_EXPORTED below. Spaces
+ // are legal in a label value but are not covered:
PromMetric#fromExportedMetric splits the scrape line
+ // on the first space, so a space defeats the harness, not the exporter.
+ private static final String OPEN_STRUCT_KEY = "clicks.v2$promo\"code";
+ // '"' escaped to '%22'; '.', '-' and '$' pass through untouched.
+ private static final String OPEN_STRUCT_KEY_EXPORTED =
"clicks.v2$promo%22code";
+ private static final String LABEL_KEY_COLUMN = "column";
+ private static final String LABEL_KEY_KEY = "key";
+
+ private static final List<ServerGauge> GAUGES_ACCEPTING_OPEN_STRUCT_COLUMN =
+ List.of(ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DENSE_KEY_COUNT,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_SPARSE_KEY_COUNT,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_COUNT,
ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DOC_COUNT);
+
+ private static final List<ServerMeter> METERS_ACCEPTING_OPEN_STRUCT_COLUMN =
+ List.of(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES);
+
+ private static final List<String> TABLENAME_TABLETYPE_COLUMN =
+ List.of(ExportedLabelKeys.TABLE, ExportedLabelValues.TABLENAME,
ExportedLabelKeys.TABLETYPE,
+ ExportedLabelValues.TABLETYPE_REALTIME, LABEL_KEY_COLUMN,
OPEN_STRUCT_COLUMN);
+
+ private static final List<String> TABLENAME_TABLETYPE_COLUMN_KEY =
+ List.of(ExportedLabelKeys.TABLE, ExportedLabelValues.TABLENAME,
ExportedLabelKeys.TABLETYPE,
+ ExportedLabelValues.TABLETYPE_REALTIME, LABEL_KEY_COLUMN,
OPEN_STRUCT_COLUMN, LABEL_KEY_KEY,
+ OPEN_STRUCT_KEY_EXPORTED);
+
// pinot.mse.* metrics share the role-agnostic prefix and must be exported
from every JVM role
// that registers MseMetrics; on server JVMs this exercises the server.yml
catch-all rule.
private static final String EXPORTED_MSE_METRIC_PREFIX = "pinot_mse_";
@@ -118,6 +156,9 @@ public abstract class ServerPrometheusMetricsTest extends
PinotPrometheusMetrics
} else if (METERS_ACCEPTING_RAW_TABLE_NAMES.contains(serverMeter)) {
addMeterWithLabels(serverMeter, ExportedLabelValues.TABLENAME);
assertMeterExportedCorrectly(serverMeter.getMeterName(),
ExportedLabels.TABLENAME);
+ } else if (METERS_ACCEPTING_OPEN_STRUCT_COLUMN.contains(serverMeter)) {
+ _serverMetrics.addMeteredTableValue(TABLE_NAME_WITH_TYPE,
OPEN_STRUCT_COLUMN, serverMeter, 4L);
+ assertMeterExportedCorrectly(serverMeter.getMeterName(),
TABLENAME_TABLETYPE_COLUMN);
} else {
//we pass tableNameWithType to all remaining meters
addMeterWithLabels(serverMeter, TABLE_NAME_WITH_TYPE);
@@ -143,6 +184,14 @@ public abstract class ServerPrometheusMetricsTest extends
PinotPrometheusMetrics
} else if (GAUGES_ACCEPTING_RAW_TABLE_NAME.contains(serverGauge)) {
addGaugeWithLabels(serverGauge, ExportedLabelValues.TABLENAME);
assertGaugeExportedCorrectly(serverGauge.getGaugeName(),
ExportedLabels.TABLENAME, EXPORTED_METRIC_PREFIX);
+ } else if (serverGauge ==
ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT) {
+ _serverMetrics.setOrUpdateTableGauge(TABLE_NAME_WITH_TYPE,
+ OpenStructNaming.metricKey(OPEN_STRUCT_COLUMN, OPEN_STRUCT_KEY),
serverGauge, 100L);
+ assertGaugeExportedCorrectly(serverGauge.getGaugeName(),
TABLENAME_TABLETYPE_COLUMN_KEY,
+ EXPORTED_METRIC_PREFIX);
+ } else if (GAUGES_ACCEPTING_OPEN_STRUCT_COLUMN.contains(serverGauge)) {
+ _serverMetrics.setOrUpdateTableGauge(TABLE_NAME_WITH_TYPE,
OPEN_STRUCT_COLUMN, serverGauge, 100L);
+ assertGaugeExportedCorrectly(serverGauge.getGaugeName(),
TABLENAME_TABLETYPE_COLUMN, EXPORTED_METRIC_PREFIX);
} else {
addGaugeWithLabels(serverGauge, TABLE_NAME_WITH_TYPE);
assertGaugeExportedCorrectly(serverGauge.getGaugeName(),
ExportedLabels.TABLENAME_TABLETYPE,
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index d9156565e49..681c546d418 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -427,7 +427,8 @@ public class MutableSegmentImpl implements MutableSegment {
if (dataType == DataType.OPEN_STRUCT && fieldSpec instanceof
ComplexFieldSpec) {
IndexConfig openStructConfig =
indexConfigs.getConfig(StandardIndexes.openStruct());
if (openStructConfig instanceof OpenStructIndexConfig &&
openStructConfig.isEnabled()) {
- MutableOpenStructIndex openStructIndex = new
MutableOpenStructIndex(column, (ComplexFieldSpec) fieldSpec,
+ MutableOpenStructIndex openStructIndex = new
MutableOpenStructIndex(column, _realtimeTableName,
+ (ComplexFieldSpec) fieldSpec,
(OpenStructIndexConfig) openStructConfig, _memoryManager,
_capacity);
mutableIndexes.put(StandardIndexes.openStruct(), openStructIndex);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
index 00419a41907..2bdbf8a0e71 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
@@ -87,6 +88,7 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
private final File _indexDir;
private final String _columnName;
+ private final String _tableNameWithType;
private final Map<String, FieldSpec> _childFieldSpecs;
private final OpenStructIndexConfig _config;
private final int _maxDenseKeys;
@@ -95,18 +97,20 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
private final Map<String, RoaringBitmap> _presenceBitmaps = new HashMap<>();
private final Map<String, List<Object>> _values = new HashMap<>();
private final Map<String, DataType> _inferredTypes = new HashMap<>();
+ private final Map<String, Long> _coercionFailuresPerKey = new HashMap<>();
+ private final Map<String, Long> _inferenceFailuresPerKey = new HashMap<>();
private int _numDocs;
- private int _coercionFailures;
// Resolved at seal time
@Nullable
private Set<String> _resolvedDenseKeys;
private final Map<String, PropertiesConfiguration>
_materializedColumnMetadata = new LinkedHashMap<>();
- public OpenStructColumnSplitter(File indexDir, String columnName, FieldSpec
fieldSpec,
+ public OpenStructColumnSplitter(File indexDir, String columnName, String
tableNameWithType, FieldSpec fieldSpec,
OpenStructIndexConfig config) {
_indexDir = indexDir;
_columnName = columnName;
+ _tableNameWithType = tableNameWithType;
_config = config;
_maxDenseKeys = config.getMaxDenseKeys();
@@ -197,12 +201,34 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
continue;
}
FieldSpec keySpec = _childFieldSpecs.get(key);
- DataType valueType = keySpec != null
- ? keySpec.getDataType()
- : _inferredTypes.computeIfAbsent(key, k -> {
- DataType inferred =
OpenStructTypeInference.inferDataType(rawValue);
- return inferred != null ? inferred : DataType.STRING;
- });
+ DataType valueType;
+ if (keySpec != null) {
+ valueType = keySpec.getDataType();
+ } else {
+ DataType established = _inferredTypes.get(key);
+ if (established != null && established != DataType.STRING) {
+ // Sticky: a key already resolved to a non-STRING type can't flip
later, so skip
+ // inference entirely -- matches MutableOpenStructIndex's fast
path. An unmappable
+ // value here is a coercion failure below, not a fresh inference
decision; overriding
+ // valueType to STRING per-row would desync it from _inferredTypes
and corrupt _values
+ // with a mix of types for one key.
+ valueType = established;
+ } else {
+ // Resolve per value rather than only on first sighting: the key's
inferred type is
+ // cached, so folding the counter into a computeIfAbsent would
record one failure per
+ // key no matter how many values actually took the STRING fallback.
+ DataType inferred =
OpenStructTypeInference.inferDataType(rawValue);
+ if (inferred == null) {
+ valueType = DataType.STRING;
+ _inferenceFailuresPerKey.merge(key, 1L, Long::sum);
+ } else {
+ // established is STRING here (or null): once a key falls back
to STRING it stays
+ // STRING even if a later value would infer cleanly on its own.
+ valueType = established != null ? established : inferred;
+ }
+ _inferredTypes.putIfAbsent(key, valueType);
+ }
+ }
if (!_presenceBitmaps.containsKey(key)) {
_presenceBitmaps.put(key, new RoaringBitmap());
_values.put(key, new ArrayList<>());
@@ -214,7 +240,7 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
PinotDataType destType =
ColumnDataType.fromDataTypeSV(valueType.getStoredType()).toPinotDataType();
coerced = destType.convert(rawValue, sourceType);
} catch (Exception e) {
- _coercionFailures++;
+ _coercionFailuresPerKey.merge(key, 1L, Long::sum);
_presenceBitmaps.get(key).remove(_numDocs);
continue;
}
@@ -246,17 +272,74 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
writeSparseJsonColumn(sparseKeys);
}
- if (_coercionFailures > 0) {
- LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures", _columnName, _coercionFailures);
- ServerMetrics serverMetrics = ServerMetrics.get();
- if (serverMetrics != null) {
-
serverMetrics.addMeteredGlobalValue(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
_coercionFailures);
- }
+ long totalCoercionFailures = sumValues(_coercionFailuresPerKey);
+ if (totalCoercionFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures across {} keys",
+ _columnName, totalCoercionFailures, _coercionFailuresPerKey.size());
+ // The key space is user-controlled, so the per-key breakdown is
DEBUG-only.
+ LOGGER.debug("OPEN_STRUCT '{}': full coercion failure counts: {}",
_columnName, _coercionFailuresPerKey);
+ }
+ long totalInferenceFailures = sumValues(_inferenceFailuresPerKey);
+ if (totalInferenceFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': {} values across {} keys fell back to
STRING after type inference failed",
+ _columnName, totalInferenceFailures,
_inferenceFailuresPerKey.size());
+ LOGGER.debug("OPEN_STRUCT '{}': full inference failure counts: {}",
_columnName, _inferenceFailuresPerKey);
}
+ emitMetrics(sparseKeys.size(), totalCoercionFailures,
totalInferenceFailures);
emitParentColumnMetadata(sparseKeys);
}
+ private static long sumValues(Map<String, Long> counts) {
+ return counts.values().stream().mapToLong(Long::longValue).sum();
+ }
+
+ private void emitMetrics(int sparseKeyCount, long totalCoercionFailures,
long totalInferenceFailures) {
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics == null || _numDocs == 0) {
+ return;
+ }
+
+ if (totalCoercionFailures > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, _columnName,
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
totalCoercionFailures);
+ }
+ if (totalInferenceFailures > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, _columnName,
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES,
totalInferenceFailures);
+ }
+
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, _columnName,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DENSE_KEY_COUNT,
_resolvedDenseKeys.size());
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, _columnName,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_SPARSE_KEY_COUNT, sparseKeyCount);
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, _columnName,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_COUNT,
_presenceBitmaps.size());
+ // Denominator for the per-key fill rate. Emitted as a raw count rather
than folding the ratio into a
+ // single percentage gauge: integer division truncates a key present in a
handful of docs to 0, which
+ // is indistinguishable from no data and is exactly the case worth
alerting on.
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, _columnName,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DOC_COUNT, _numDocs);
+
+ if (_config.isPerKeyMetricsEnabled()) {
+ // Emit for every key in the segment. Registry entries follow the
ingested key space;
+ // table deletion can only sweep keys recoverable from denseKeys.
+ _presenceBitmaps.forEach((key, presence) ->
serverMetrics.setOrUpdateTableGauge(_tableNameWithType,
+ OpenStructNaming.metricKey(_columnName, key),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT,
presence.getCardinality()));
+ } else {
+ // Emit only for configured dense keys — bounded by the table config.
+ for (String key : _config.getDenseKeys()) {
+ RoaringBitmap presence = _presenceBitmaps.get(key);
+ if (presence != null) {
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType,
+ OpenStructNaming.metricKey(_columnName, key),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT,
presence.getCardinality());
+ }
+ }
+ }
+ }
+
@Override
public void close()
throws IOException {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
index f000232e0c0..648f0dd5ee9 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
@@ -57,6 +57,7 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
private static final Logger LOGGER =
LoggerFactory.getLogger(MutableOpenStructIndex.class);
private final String _openStructColumn;
+ private final String _tableNameWithType;
private final OpenStructIndexConfig _config;
private final Map<String, FieldSpec> _childFieldSpecs;
private final PinotDataBufferMemoryManager _memoryManager;
@@ -65,9 +66,10 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
// Volatile for lock-free reader access; writer always holds the
consuming-thread lock.
private volatile Map<String, MutableKeyColumn> _keyColumns = new HashMap<>();
- public MutableOpenStructIndex(String openStructColumn, ComplexFieldSpec
fieldSpec,
+ public MutableOpenStructIndex(String openStructColumn, String
tableNameWithType, ComplexFieldSpec fieldSpec,
OpenStructIndexConfig config, PinotDataBufferMemoryManager
memoryManager, int capacity) {
_openStructColumn = openStructColumn;
+ _tableNameWithType = tableNameWithType;
_config = config;
_memoryManager = memoryManager;
_capacity = capacity;
@@ -108,10 +110,7 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
// build, so no key is dropped during consumption.
// Resolve stored type and coerce BEFORE allocating a column so a
first-row coercion failure
// does not allocate a column that was never usable.
- DataType resolvedType = resolveStoredType(key, rawValue);
- if (resolvedType == null) {
- continue;
- }
+ DataType resolvedType = resolveStoredType(key, rawValue, null);
Object coerced = tryCoerce(key, rawValue, resolvedType);
if (coerced == null) {
continue;
@@ -121,7 +120,9 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
continue;
}
- DataType storedType = keyCol.getStoredType();
+ // Re-resolve against the established type so a later unmappable value
on a STRING key is
+ // metered too; for any other established type this is a no-op returning
that type.
+ DataType storedType = resolveStoredType(key, rawValue,
keyCol.getStoredType());
Object coerced = tryCoerce(key, rawValue, storedType);
if (coerced == null) {
continue;
@@ -130,29 +131,42 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
}
}
- /// Resolves the stored type for a key without allocating any state. Returns
null when the type
- /// cannot be inferred (caller should skip the entry).
- @Nullable
- private DataType resolveStoredType(String key, Object rawValue) {
+ /// Resolves the stored type for a key without allocating any state, and
meters a value that took
+ /// the STRING fallback. `establishedType` is the key's already-resolved
stored type, or `null` on
+ /// first sighting.
+ ///
+ /// The fallback rule (unmappable value → STRING) must match the sealed
build path
+ /// ([OpenStructColumnSplitter#addMap]) so a value reads the same before and
after seal.
+ private DataType resolveStoredType(String key, Object rawValue, @Nullable
DataType establishedType) {
FieldSpec spec = _childFieldSpecs.get(key);
- DataType valueType;
if (spec != null) {
- valueType = spec.getDataType();
- } else {
- valueType = OpenStructTypeInference.inferDataType(rawValue);
- if (valueType == null) {
+ return spec.getDataType().getStoredType();
+ }
+ if (establishedType != null && establishedType != DataType.STRING) {
+ return establishedType;
+ }
+ DataType inferred = OpenStructTypeInference.inferDataType(rawValue);
+ if (inferred == null) {
+ if (establishedType == null) {
LOGGER.warn("OPEN_STRUCT '{}': could not infer DataType for key '{}'
from value of class '{}'."
- + " Dropping the entry.",
+ + " Falling back to STRING.",
_openStructColumn, key, rawValue.getClass().getName());
- return null;
}
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics != null) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn,
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1);
+ }
+ return DataType.STRING;
}
- return valueType.getStoredType();
+ return establishedType != null ? establishedType : inferred;
}
- /// Coerces rawValue to storedType. Returns null on failure (logged at
WARN); the caller drops
- /// the entry. Note: a successful coerce of a "null"-shaped raw value would
also return null —
- /// but callers gate on rawValue != null before reaching here.
+ /// Coerces rawValue to storedType. Returns null on failure; the caller
drops the entry. Failures
+ /// are reported through [ServerMeter#OPEN_STRUCT_TYPE_COERCION_FAILURES]
rather than a log line,
+ /// because this runs per value on the consuming path. Note: a successful
coerce of a
+ /// "null"-shaped raw value would also return null — but callers gate on
rawValue != null before
+ /// reaching here.
@Nullable
private Object tryCoerce(String key, Object rawValue, DataType storedType) {
try {
@@ -162,7 +176,9 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
} catch (Exception e) {
ServerMetrics serverMetrics = ServerMetrics.get();
if (serverMetrics != null) {
-
serverMetrics.addMeteredGlobalValue(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
1);
+ // Column-granular for the same reason as the inference meter above.
+ serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn,
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1);
}
return null;
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
index 787b383c12a..383975c1ba4 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
@@ -134,7 +134,8 @@ public class OpenStructIndexType
public ColumnarOpenStructIndexCreator
createIndexCreator(IndexCreationContext context,
OpenStructIndexConfig indexConfig) {
FieldSpec fieldSpec = context.getFieldSpec();
- return new OpenStructColumnSplitter(context.getIndexDir(),
fieldSpec.getName(), fieldSpec, indexConfig);
+ return new OpenStructColumnSplitter(context.getIndexDir(),
fieldSpec.getName(), context.getTableNameWithType(),
+ fieldSpec, indexConfig);
}
@Override
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
index d8993af7da1..abb140a8b60 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
@@ -29,6 +29,9 @@ import java.util.Map;
import java.util.Set;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.FilterContext;
import org.apache.pinot.common.request.context.predicate.EqPredicate;
@@ -48,6 +51,12 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
@@ -76,13 +85,19 @@ public class OpenStructColumnSplitterTest {
}
private OpenStructIndexConfig config(double minFillRate, int maxDenseKeys,
Set<String> denseKeys) {
- return new OpenStructIndexConfig(false, null, maxDenseKeys, denseKeys,
minFillRate, null, null);
+ return config(minFillRate, maxDenseKeys, denseKeys, false);
+ }
+
+ private OpenStructIndexConfig config(double minFillRate, int maxDenseKeys,
Set<String> denseKeys,
+ boolean perKeyMetricsEnabled) {
+ return new OpenStructIndexConfig(false, null, maxDenseKeys, denseKeys,
minFillRate, null, null,
+ perKeyMetricsEnabled);
}
@Test
public void testClassifyByFillRate()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
Map<String, Object> doc = d < 7 ? Map.of("clicks", (long) d) : Map.of();
@@ -95,7 +110,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testExplicitDenseKeysAlwaysMaterialized()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.99, -1, Set.of("rare")));
s.add(Map.of("rare", "x"), 0);
for (int d = 1; d < 100; d++) {
@@ -108,7 +123,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testRareKeyDroppedFromDense()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
s.add(Map.of("rare", "x"), 0);
for (int d = 1; d < 100; d++) {
@@ -121,7 +136,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testMaxDenseKeysCap()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.1, 1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("a", "x", "b", "y", "c", "z"), d);
@@ -133,7 +148,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testZeroDocsIsNoop()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
s.seal();
assertTrue(s.getResolvedDenseKeys().isEmpty());
@@ -142,7 +157,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testSealEmitsParentMetadataForDense()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("clicks", (long) d), d);
@@ -167,7 +182,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testDenseColumnMetadataKeysPresent()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("clicks", (long) d), d);
@@ -194,7 +209,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testSparseJsonColumnWritten()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.9, -1, null));
s.add(Map.of("rare", "x"), 0);
for (int d = 1; d < 10; d++) {
@@ -210,7 +225,7 @@ public class OpenStructColumnSplitterTest {
throws Exception {
// Regression: an untyped key whose value is a BigDecimal used to crash
seal() with
// IllegalStateException("Unsupported OPEN_STRUCT stored type for
dictionary build: BIG_DECIMAL").
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("amount", new
BigDecimal("12.34").add(BigDecimal.valueOf(d))), d);
@@ -233,7 +248,7 @@ public class OpenStructColumnSplitterTest {
// 1.0 and 1.00 are equal by compareTo but distinct by equals; they must
stay separate dictionary
// entries. Doc 2 is absent, so the default (BigDecimal.ZERO) is also
collected -> 3 distinct values.
// A compareTo-based dedup would wrongly collapse 1.0/1.00 and yield 2.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
s.add(Map.of("amount", new BigDecimal("1.0")), 0);
s.add(Map.of("amount", new BigDecimal("1.00")), 1);
@@ -254,7 +269,7 @@ public class OpenStructColumnSplitterTest {
Map<String, FieldSpec> children = Map.of(
"amount", new DimensionFieldSpec("amount", DataType.BIG_DECIMAL,
true));
ComplexFieldSpec specWithChild = new ComplexFieldSpec("metrics",
DataType.OPEN_STRUCT, true, children);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", specWithChild,
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", specWithChild,
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("amount", new BigDecimal("100.5")), d);
@@ -276,7 +291,7 @@ public class OpenStructColumnSplitterTest {
.withEncodingType(FieldConfig.EncodingType.RAW).build();
OpenStructIndexConfig cfg = new OpenStructIndexConfig(
false, null, -1, null, 0.5, List.of(rawConfig), null);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
for (int d = 0; d < 10; d++) {
s.add(Map.of("amount", new
BigDecimal("7.5").add(BigDecimal.valueOf(d))), d);
}
@@ -296,7 +311,7 @@ public class OpenStructColumnSplitterTest {
public void testBigDecimalSparseKey()
throws Exception {
// A BIG_DECIMAL key below the fill-rate threshold goes to the sparse JSON
column without crashing.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.9, -1, null));
s.add(Map.of("rare", new BigDecimal("3.14159")), 0);
for (int d = 1; d < 10; d++) {
@@ -311,7 +326,7 @@ public class OpenStructColumnSplitterTest {
throws Exception {
// Absent docs now store the standard Pinot dimension null value (INT ->
Integer.MIN_VALUE),
// so the column min reflects that default rather than the old
metric-style 0.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 5; d++) {
s.add(Map.of("clicks", 10 + d), d); // present: 10..14
@@ -338,7 +353,7 @@ public class OpenStructColumnSplitterTest {
throws Exception {
// Default keys are dictionary-encoded with an inverted index (both
default on), now written via the
// standard ForwardIndexCreator and inverted index creator.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("tag", "v" + (d % 3)), d);
@@ -364,7 +379,7 @@ public class OpenStructColumnSplitterTest {
FieldConfig rawConfig = new FieldConfig.Builder("note")
.withEncodingType(FieldConfig.EncodingType.RAW).build();
OpenStructIndexConfig cfg = new OpenStructIndexConfig(false, null, -1,
null, 0.5, List.of(rawConfig), null);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
for (int d = 0; d < 10; d++) {
s.add(Map.of("note", "n" + d), d);
}
@@ -387,7 +402,7 @@ public class OpenStructColumnSplitterTest {
JsonNode indexes = JsonUtils.stringToJsonNode("{\"range\": {}, \"bloom\":
{}}");
FieldConfig keyConfig = new
FieldConfig.Builder("clicks").withIndexes(indexes).build();
OpenStructIndexConfig cfg = new OpenStructIndexConfig(false, null, -1,
null, 0.5, List.of(keyConfig), null);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
for (int d = 0; d < 10; d++) {
s.add(Map.of("clicks", d), d);
}
@@ -412,7 +427,7 @@ public class OpenStructColumnSplitterTest {
.withIndexes(indexes)
.build();
OpenStructIndexConfig cfg = new OpenStructIndexConfig(false, null, -1,
null, 0.5, List.of(keyConfig), null);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
for (int d = 0; d < 10; d++) {
s.add(Map.of("tag", "v" + d), d);
}
@@ -424,7 +439,7 @@ public class OpenStructColumnSplitterTest {
public void testParentMetadataCarriesSparseKeyManifest()
throws Exception {
// maxDenseKeys=0 forces every key sparse regardless of fill rate.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, 0, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("region", "us", "latencyMs", (long) d), d);
@@ -446,7 +461,7 @@ public class OpenStructColumnSplitterTest {
throws Exception {
// "clicks" is present on every doc (dense); "rare" is present on one doc
(sparse). The manifest
// must list only the sparse key, not the dense one.
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
s.add(Map.of("clicks", 1L, "rare", "x"), 0);
for (int d = 1; d < 10; d++) {
@@ -467,7 +482,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testParentMetadataOmitsManifestWhenNoSparseKeys()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, -1, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("clicks", (long) d), d);
@@ -485,7 +500,7 @@ public class OpenStructColumnSplitterTest {
@Test
public void testParentMetadataManifestIncludesCommaKey()
throws Exception {
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(),
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
config(0.5, 0, null));
for (int d = 0; d < 10; d++) {
s.add(Map.of("region", "us", "weird,key", "x"), d);
@@ -505,7 +520,7 @@ public class OpenStructColumnSplitterTest {
throws Exception {
// Pins bare key form (not "$."-prefixed) — MapFilterOperator fast path
relies on this.
OpenStructIndexConfig cfg = new OpenStructIndexConfig(false, null, 0,
null, null, null, true);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
s.add(Map.of("region", "us"), 0);
s.add(Map.of(), 1);
s.add(Map.of("region", "eu"), 2);
@@ -538,7 +553,7 @@ public class OpenStructColumnSplitterTest {
public void testSparseJsonIndexAbsentByDefault()
throws Exception {
OpenStructIndexConfig cfg = new OpenStructIndexConfig(false, null, 0,
null, null, null, null);
- OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", spec(), cfg);
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(), cfg);
s.add(Map.of("region", "us"), 0);
s.add(Map.of(), 1);
s.add(Map.of("region", "eu"), 2);
@@ -548,4 +563,233 @@ public class OpenStructColumnSplitterTest {
File indexFile = new File(_tempDir, sparseCol +
V1Constants.Indexes.JSON_INDEX_FILE_EXTENSION);
assertFalse(indexFile.exists());
}
+
+ /// The inferred type of a key is cached on first sighting, so counting
inference failures inside
+ /// that caching step would record 1 failure per key no matter how many
values actually failed.
+ /// Every value that takes the STRING fallback must be counted.
+ @Test
+ public void testInferenceFailuresCountedPerValueNotPerKey()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, null));
+ // 'payload' has no child spec and a Map value, which
OpenStructTypeInference cannot map to a DataType.
+ for (int d = 0; d < 4; d++) {
+ s.add(Map.of("payload", Map.of("nested", d)), d);
+ }
+ s.seal();
+
+ verify(metrics).addMeteredTableValue("testTable_OFFLINE", "metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 4L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// A value that cannot be mapped to a DataType but lands on a key whose
type is already
+ /// established as something other than STRING is dropped by coercion, not
stored as STRING. It
+ /// must be counted once, against the coercion meter only — counting it as
an inference failure
+ /// too would record one dropped value against two meters and make the
seal-time log claim it
+ /// "fell back to STRING" when it did not.
+ @Test
+ public void testUnmappableValueOnTypedKeyCountsOnlyAsCoercionFailure()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, null));
+ // First value fixes the key's type as LONG; the next two cannot be
coerced to it.
+ s.add(Map.of("clicks", 1L), 0);
+ s.add(Map.of("clicks", Map.of("a", 1)), 1);
+ s.add(Map.of("clicks", List.of(1, 2)), 2);
+ s.seal();
+
+ verify(metrics).addMeteredTableValue("testTable_OFFLINE", "metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 2L);
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES), anyLong());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// Fill rate is emitted as two raw counts rather than one percentage gauge:
integer division
+ /// truncates a dense key present in a handful of docs to 0, which is
indistinguishable from no
+ /// data and is exactly the case worth alerting on. Pins both numerator and
denominator for a key
+ /// whose fill rate would truncate to 0.
+ @Test
+ public void testGaugesEmitRawDocCountsForTruncatingFillRate()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ // 'rare' is forced dense despite appearing in 3 of 200 docs (1.5%,
which truncates to 0%).
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, Set.of("rare")));
+ for (int d = 0; d < 200; d++) {
+ s.add(d < 3 ? Map.of("rare", (long) d, "host", "h") : Map.of("host",
"h"), d);
+ }
+ s.classify();
+ s.seal();
+
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DOC_COUNT, 200L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
+ OpenStructNaming.materializedColumnName("metrics", "rare"),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 3L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_COUNT, 2L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DENSE_KEY_COUNT, 2L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_SPARSE_KEY_COUNT, 0L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// Default: flag off, no configured denseKeys → no per-key gauge at all.
+ @Test
+ public void testPerKeyGaugeOffByDefaultWithNoDenseKeys()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, null));
+ for (int d = 0; d < 10; d++) {
+ s.add(Map.of("host", "h", "clicks", (long) d), d);
+ }
+ s.classify();
+ s.seal();
+
+ verify(metrics, never()).setOrUpdateTableGauge(anyString(), anyString(),
+ eq(ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT), anyLong());
+ // Column-level gauges still emit.
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_DENSE_KEY_COUNT, 2L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// Flag on: every discovered key emits, including sparse ones.
+ @Test
+ public void testPerKeyGaugeCoversAllKeysWhenEnabled()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, null, true));
+ // 'host' fills every doc (dense); 'rare' fills 2 of 10 (sparse).
+ for (int d = 0; d < 10; d++) {
+ s.add(d < 2 ? Map.of("host", "h", "rare", (long) d) : Map.of("host",
"h"), d);
+ }
+ s.classify();
+ s.seal();
+
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
OpenStructNaming.metricKey("metrics", "host"),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 10L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
OpenStructNaming.metricKey("metrics", "rare"),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 2L);
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE", "metrics",
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_SPARSE_KEY_COUNT, 1L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// A configured key cut from the dense set by the maxDenseKeys cap still
reports its doc count -- a
+ /// configured key that did not earn a materialized column is the case worth
seeing. This is the branch
+ /// where iterating the keys present in the segment diverges from iterating
`_resolvedDenseKeys`.
+ @Test
+ public void testPerKeyGaugeCoversConfiguredKeyCutByMaxDenseKeysCap()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ // Both keys qualify on fill rate (100%), but the cap admits only one to
the dense set.
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, 1, Set.of("alpha", "beta")));
+ for (int d = 0; d < 10; d++) {
+ s.add(Map.of("alpha", (long) d, "beta", (long) d), d);
+ }
+ // Which of the two wins the single slot depends on Set.of iteration
order, so assert the count
+ // rather than the identity; the gauge assertions below hold either way.
+ assertEquals(s.classify().size(), 1);
+ s.seal();
+
+ for (String key : List.of("alpha", "beta")) {
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
+ OpenStructNaming.materializedColumnName("metrics", key),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 10L);
+ }
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// A configured key that never appears in the segment has no presence
bitmap and is skipped rather
+ /// than reported as 0, which would be indistinguishable from a key that was
ingested into no docs.
+ @Test
+ public void testPerKeyGaugeSkipsConfiguredKeyAbsentFromSegment()
+ throws Exception {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, Set.of("host", "never-ingested")));
+ for (int d = 0; d < 10; d++) {
+ s.add(Map.of("host", "h"), d);
+ }
+ s.classify();
+ s.seal();
+
+ verify(metrics, never()).setOrUpdateTableGauge(
+ eq("testTable_OFFLINE"), eq(OpenStructNaming.metricKey("metrics",
"never-ingested")),
+ eq(ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT), anyLong());
+ // The configured key that was ingested still reports, so the assertion
above is about
+ // absence, not about the gauge being off entirely.
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
OpenStructNaming.metricKey("metrics", "host"),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 10L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// Pins the emission path to `metricKey`, not `materializedColumnName`.
OPEN_STRUCT keys come from
+ /// user JSON, and a '"' in one is backslash-escaped by ObjectName.quote on
the way to JMX, which stops
+ /// the exported name matching the per-key scrape rule at all -- the key
silently loses its column/key
+ /// labels. Every other per-key test above uses a clean key, for which the
two helpers agree, so this is
+ /// the only case here that fails if the call site regresses.
+ @Test
+ public void testPerKeyGaugeEscapesKeyForMetricName()
+ throws Exception {
+ String rawKey = "promo\"code";
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ OpenStructColumnSplitter s = new OpenStructColumnSplitter(_tempDir,
"metrics", "testTable_OFFLINE", spec(),
+ config(0.5, -1, Set.of(rawKey)));
+ for (int d = 0; d < 10; d++) {
+ s.add(Map.of("host", "h", rawKey, (long) d), d);
+ }
+ s.classify();
+ s.seal();
+
+ verify(metrics).setOrUpdateTableGauge("testTable_OFFLINE",
+ OpenStructNaming.metricKey("metrics", rawKey),
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT, 10L);
+ // The unescaped form is what the old call site produced; it must not be
emitted.
+ verify(metrics, never()).setOrUpdateTableGauge(
+ eq("testTable_OFFLINE"),
eq(OpenStructNaming.materializedColumnName("metrics", rawKey)),
+ eq(ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT), anyLong());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructDataSourceTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructDataSourceTest.java
index 99388c083e4..271490e850b 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructDataSourceTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructDataSourceTest.java
@@ -66,7 +66,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testGetDataSourcePerKey()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
@@ -80,7 +80,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testGetDataSourceForUnknownKey()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 0);
assertNull(ds.getDataSource("missing"));
@@ -91,7 +91,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testGetDataSourcesReturnsAllKeys()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L, "country", "US"));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
@@ -103,7 +103,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testGetDataSourceIsMemoisedPerKey()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
@@ -117,7 +117,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testMemoisedDataSourceStaysCorrectAsIngestionContinues()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
@@ -137,7 +137,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testMemoisedNullBitmapReportsAbsentDocs()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("country", "US"));
idx.index(1, Map.of("clicks", 5L));
@@ -152,7 +152,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testKeyCreatedAfterFirstLookupIsPickedUp()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
assertNull(ds.getDataSource("clicks"));
@@ -164,7 +164,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testDictionaryReservesDefaultNullValueAtDictIdZero()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 1);
@@ -186,7 +186,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testLastIndexedDocIdWatermark()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
idx.index(1, Map.of("other", 1L));
@@ -201,7 +201,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testForwardIndexSafeForAbsentTailDocs()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 3000)) {
idx.index(0, Map.of("clicks", 5L));
idx.index(3, Map.of("clicks", 9L));
@@ -245,7 +245,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testInvertedIndexFoldsAbsentDocsIntoDefaultPostings()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
// Present in docs 0 (5L) and 3 (9L) of 6; docs 1,2,4,5 never see the
key.
idx.index(0, Map.of("clicks", 5L));
@@ -266,7 +266,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testInvertedIndexFoldsAbsentDocsWithExplicitDefaultWrite()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
// doc 0 writes a real value; doc 3 explicitly writes the LONG default
itself.
idx.index(0, Map.of("clicks", 5L));
@@ -284,7 +284,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testKeyDictionaryExactWhenPartiallyPresent()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableOpenStructDataSource ds = new MutableOpenStructDataSource(spec(),
idx, 3);
@@ -297,7 +297,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testKeyDictionaryNotExactWhenFullyPresentAndDefaultUnobserved()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
idx.index(1, Map.of("clicks", 7L));
@@ -310,7 +310,7 @@ public class MutableOpenStructDataSourceTest {
@Test
public void testKeyDictionaryExactWhenFullyPresentAndDefaultObserved()
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
idx.index(1, Map.of("clicks", Long.MIN_VALUE));
@@ -332,7 +332,7 @@ public class MutableOpenStructDataSourceTest {
children.put("clicks", new DimensionFieldSpec("clicks", DataType.LONG,
true, 0L));
ComplexFieldSpec specWithCustomDefault =
new ComplexFieldSpec("metrics", DataType.OPEN_STRUCT, true, children);
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
specWithCustomDefault,
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", specWithCustomDefault,
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("clicks", 5L));
MutableKeyColumn col = idx.getKeyColumn("clicks");
@@ -357,7 +357,7 @@ public class MutableOpenStructDataSourceTest {
@Test(dataProvider = "storedTypes")
public void testDefaultReservationPerStoredType(DataType storedType, Object
value)
throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, 100)) {
idx.index(0, Map.of("k", value));
MutableKeyColumn col = idx.getKeyColumn("k");
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
index fc031f04df0..492cbeac33e 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
@@ -21,6 +21,8 @@ package
org.apache.pinot.segment.local.segment.index.openstruct;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
@@ -30,6 +32,14 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
@@ -58,7 +68,7 @@ public class MutableOpenStructIndexTest {
public void testAddAndGetKeys()
throws IOException {
try (MutableOpenStructIndex idx = new MutableOpenStructIndex(
- "metrics", openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr,
1000)) {
+ "metrics", "testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 1000)) {
idx.index(0, Map.of("clicks", 42L, "impressions", 100L));
idx.index(1, Map.of("clicks", 7L, "revenue", "1.5"));
@@ -75,7 +85,7 @@ public class MutableOpenStructIndexTest {
public void testIndexNullIsNoop()
throws IOException {
try (MutableOpenStructIndex idx = new MutableOpenStructIndex(
- "metrics", openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr,
1000)) {
+ "metrics", "testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 1000)) {
idx.index(0, null);
@@ -88,7 +98,7 @@ public class MutableOpenStructIndexTest {
public void testFillRateTracking()
throws IOException {
try (MutableOpenStructIndex idx = new MutableOpenStructIndex(
- "metrics", openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr,
1000)) {
+ "metrics", "testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 1000)) {
for (int docId = 0; docId < 10; docId++) {
if (docId < 7) {
@@ -111,7 +121,7 @@ public class MutableOpenStructIndexTest {
throws IOException {
// No childFieldSpecs — type inference from rawValue
ComplexFieldSpec spec = new ComplexFieldSpec("metrics",
DataType.OPEN_STRUCT, true, Map.of());
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
spec,
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", spec,
OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
idx.index(0, java.util.Map.of("clicks", 5L));
assertEquals(idx.getKeyColumn("clicks").getStoredType(), DataType.LONG);
@@ -122,7 +132,7 @@ public class MutableOpenStructIndexTest {
@Test
public void testImplementsOpenStructIndexReader() throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
openStructSpec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
assertTrue(idx instanceof
org.apache.pinot.segment.spi.index.reader.OpenStructIndexReader);
}
@@ -130,7 +140,7 @@ public class MutableOpenStructIndexTest {
@Test
public void testGetIndexesReturnsForwardIndexForMaterializedKey() throws
Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
openStructSpec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
idx.index(0, Map.of("clicks", 5L));
Map<org.apache.pinot.segment.spi.index.IndexType,
org.apache.pinot.segment.spi.index.IndexReader> indexes =
@@ -141,7 +151,7 @@ public class MutableOpenStructIndexTest {
@Test
public void testGetIndexesUnknownKeyReturnsEmpty() throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
openStructSpec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
assertTrue(idx.getIndexes("missing").isEmpty());
}
@@ -149,7 +159,7 @@ public class MutableOpenStructIndexTest {
@Test
public void testGetColumnMetadataReturnsKeyMetadata() throws Exception {
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
openStructSpec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
idx.index(0, Map.of("clicks", 5L));
assertNotNull(idx.getColumnMetadata("clicks"));
@@ -157,4 +167,50 @@ public class MutableOpenStructIndexTest {
assertNull(idx.getColumnMetadata("absent"));
}
}
+
+ /// Both failure meters are keyed on the OPEN_STRUCT column, never on the
per-key materialized
+ /// name. These fire on malformed input, so an id-like key would otherwise
mint one meter per id,
+ /// and meters are never removed from the registry.
+ @Test
+ public void testFailureMetersAreKeyedOnColumnNotKey()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
+ OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ // Unmappable value on a fresh key: falls back to STRING and meters an
inference failure.
+ idx.index(0, Map.of("req-42", Map.of("a", 1)));
+ // Unmappable value on a key already typed LONG: dropped by coercion,
metered there only.
+ idx.index(1, Map.of("clicks", 5L));
+ idx.index(2, Map.of("clicks", Map.of("a", 1)));
+
+ verify(metrics).addMeteredTableValue("testTable_REALTIME", "metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1L);
+ verify(metrics).addMeteredTableValue("testTable_REALTIME", "metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1L);
+ verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$req-42"), any(), anyLong());
+ verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$clicks"), any(), anyLong());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// A later unmappable value on a key whose type fell back to STRING is
stored as its serialized
+ /// form, so it is metered every time — not just on the first sighting that
established the type.
+ @Test
+ public void testInferenceFailuresMeteredPerValueOnStringFallbackKey()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
+ OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ for (int docId = 0; docId < 3; docId++) {
+ idx.index(docId, Map.of("payload", Map.of("a", docId)));
+ }
+ verify(metrics, times(3)).addMeteredTableValue("testTable_REALTIME",
"metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1L);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructConsumingSealedParityTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructConsumingSealedParityTest.java
index db8ce5c48f1..1495973f410 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructConsumingSealedParityTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructConsumingSealedParityTest.java
@@ -116,7 +116,7 @@ public class OpenStructConsumingSealedParityTest {
// --- Consuming side ---
List<Object> consumingValues;
MutableRoaringBitmap consumingDefaultDocIds;
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex(METRICS,
spec(),
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex(METRICS,
"testTable_REALTIME", spec(),
OpenStructIndexConfig.DEFAULT, _mm, NUM_DOCS)) {
for (int docId = 0; docId < NUM_DOCS; docId++) {
idx.index(docId, metricsForDoc(docId));
@@ -201,6 +201,82 @@ public class OpenStructConsumingSealedParityTest {
}
}
+ /// A key with no declared child spec whose value is a Map or a List:
`OpenStructTypeInference`
+ /// maps neither to a Pinot DataType, so both tiers must fall back to STRING
and store the
+ /// serialized form. Before this was aligned, the consuming tier dropped the
entry while the
+ /// sealed tier kept it, so the same row read differently either side of the
seal boundary (and
+ /// the REALTIME and OFFLINE halves of a hybrid table disagreed).
+ @Test
+ public void testConsumingMatchesSealedForUninferrableValues()
+ throws Exception {
+ String nestedKey = "payload";
+ Map<Integer, Object> raw = new HashMap<>();
+ raw.put(0, Map.of("b", 2, "a", 1));
+ raw.put(1, List.of(1, 2, 3));
+ for (int docId = 2; docId < NUM_DOCS; docId++) {
+ raw.put(docId, "plain-" + docId);
+ }
+
+ // 'payload' is deliberately absent from the child specs so that type
inference runs for it.
+ ComplexFieldSpec inferredSpec = new ComplexFieldSpec(METRICS,
FieldSpec.DataType.OPEN_STRUCT, true,
+ Map.of("host", new DimensionFieldSpec("host",
FieldSpec.DataType.STRING, true)));
+
+ List<Object> consumingValues;
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex(METRICS,
"testTable_REALTIME", inferredSpec,
+ OpenStructIndexConfig.DEFAULT, _mm, NUM_DOCS)) {
+ for (int docId = 0; docId < NUM_DOCS; docId++) {
+ idx.index(docId, Map.of("host", "host-" + docId, nestedKey,
raw.get(docId)));
+ }
+ MutableOpenStructDataSource ds = new
MutableOpenStructDataSource(inferredSpec, idx, NUM_DOCS);
+ DataSource payload = ds.getDataSource(nestedKey);
+ assertNotNull(payload, "uninferrable key must still be materialized on
the consuming side");
+ consumingValues = readAllValues(payload);
+ }
+
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("testOpenStructInferParity")
+ .addField(inferredSpec)
+ .build();
+ OpenStructIndexConfig osConfig =
+ new OpenStructIndexConfig(false, null, -1, Set.of(nestedKey, "host"),
0.5, List.of(), null);
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("open_struct", JsonUtils.objectToJsonNode(osConfig));
+ FieldConfig metricsCfg = new
FieldConfig.Builder(METRICS).withIndexes(indexes).build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("testOpenStructInferParity")
+ .setFieldConfigList(List.of(metricsCfg)).build();
+
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig,
schema);
+ config.setOutDir(TMP_DIR.getAbsolutePath());
+ config.setSegmentName("testSegmentInferParity");
+
+ List<GenericRow> rows = new ArrayList<>(NUM_DOCS);
+ for (int docId = 0; docId < NUM_DOCS; docId++) {
+ GenericRow row = new GenericRow();
+ row.putValue(METRICS, Map.of("host", "host-" + docId, nestedKey,
raw.get(docId)));
+ rows.add(row);
+ }
+
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+
+ ImmutableSegment sealed =
ImmutableSegmentLoader.load(driver.getOutputDirectory(), ReadMode.mmap);
+ try {
+ OpenStructDataSource sealedMetrics = (OpenStructDataSource)
sealed.getDataSource(METRICS);
+ DataSource sealedPayload = sealedMetrics.getDataSource(nestedKey);
+ assertNotNull(sealedPayload, "uninferrable key must still be
materialized on the sealed side");
+ List<Object> sealedValues = readAllValues(sealedPayload);
+
+ assertEquals(consumingValues, sealedValues);
+ // Pin the serialized form so a change in either tier's fallback is
caught, not just divergence.
+ // MapUtils.toString sorts by key, so the input order {"b","a"} must
come back out as {"a","b"}.
+ assertEquals(consumingValues.get(0), "{\"a\":1,\"b\":2}");
+ assertEquals(consumingValues.get(1), "[1, 2, 3]");
+ assertEquals(consumingValues.get(2), "plain-2");
+ } finally {
+ sealed.destroy();
+ }
+ }
+
private static long min(List<Object> values) {
return values.stream().mapToLong(v -> ((Number)
v).longValue()).min().orElseThrow();
}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
index cf7498b0dee..9312477a3be 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
@@ -18,11 +18,14 @@
*/
package org.apache.pinot.server.starter.helix;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.NotificationContext;
import org.apache.helix.messaging.handling.HelixTaskResult;
import org.apache.helix.messaging.handling.MessageHandler;
@@ -43,6 +46,13 @@ import org.apache.pinot.common.metrics.ServerTimer;
import org.apache.pinot.core.data.manager.InstanceDataManager;
import org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigsUtil;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.spi.config.table.IndexConfig;
+import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.OpenStructNaming;
+import org.apache.pinot.spi.data.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -169,6 +179,9 @@ public class SegmentMessageHandlerFactory implements
MessageHandlerFactory {
public HelixTaskResult handleMessage() {
HelixTaskResult helixTaskResult = new HelixTaskResult();
_logger.info("Handling table deletion message: {}", _message);
+ // Resolved before the table goes away: these gauge keys come from the
table config, and the only
+ // in-memory copy of it lives on the table data manager that deleteTable
is about to discard.
+ List<String> openStructMetricKeys = resolveOpenStructMetricKeys();
try {
long deletionTimeMs = _message.getCreateTimeStamp();
if (deletionTimeMs <= 0) {
@@ -188,6 +201,12 @@ public class SegmentMessageHandlerFactory implements
MessageHandlerFactory {
Arrays.stream(ServerGauge.values())
.filter(g -> !g.isGlobal())
.forEach(g -> _metrics.removeTableGauge(_tableNameWithType, g));
+ // OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT registers as
<gauge>.<table>.<column>$<key>, so the sweep above --
+ // which composes only the unkeyed <gauge>.<table> -- cannot reach it.
Only the configured keys are
+ // recoverable here; see openStructMetricKeys for the keys this misses.
+ openStructMetricKeys.forEach(
+ metricKey -> _metrics.removeTableGauge(_tableNameWithType,
metricKey,
+ ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT));
Arrays.stream(ServerTimer.values())
.filter(t -> !t.isGlobal())
.forEach(t -> _metrics.removeTableTimer(_tableNameWithType, t));
@@ -199,6 +218,41 @@ public class SegmentMessageHandlerFactory implements
MessageHandlerFactory {
}
return helixTaskResult;
}
+
+ private List<String> resolveOpenStructMetricKeys() {
+ try {
+ TableDataManager tableDataManager =
_instanceDataManager.getTableDataManager(_tableNameWithType);
+ if (tableDataManager == null) {
+ return List.of();
+ }
+ Pair<TableConfig, Schema> configAndSchema =
tableDataManager.getCachedTableConfigAndSchema();
+ return configAndSchema == null ? List.of()
+ : openStructMetricKeys(configAndSchema.getLeft(),
configAndSchema.getRight());
+ } catch (Exception e) {
+ // Metric bookkeeping must never block table deletion.
+ _logger.warn("Could not resolve OPEN_STRUCT gauge keys for table: {};
its per-key gauges may survive until "
+ + "the next restart", _tableNameWithType, e);
+ return List.of();
+ }
+ }
+ }
+
+ /// The `<column>$<key>` metric keys of the per-key OPEN_STRUCT gauges
recoverable from the given
+ /// table's config — one per configured `denseKeys` entry. Complete when
`perKeyMetricsEnabled` is
+ /// off; a subset when on, since discovered keys exist only in ingested data
and cannot be named at
+ /// deletion time. Gauges for discovered keys survive until the server
restarts.
+ @VisibleForTesting
+ static List<String> openStructMetricKeys(TableConfig tableConfig, Schema
schema) {
+ List<String> metricKeys = new ArrayList<>();
+ FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig,
schema).forEach((column, indexConfigs) -> {
+ IndexConfig openStructConfig =
indexConfigs.getConfig(StandardIndexes.openStruct());
+ if (openStructConfig instanceof OpenStructIndexConfig) {
+ for (String key : ((OpenStructIndexConfig)
openStructConfig).getDenseKeys()) {
+ metricKeys.add(OpenStructNaming.metricKey(column, key));
+ }
+ }
+ });
+ return metricKeys;
}
private class ForceCommitMessageHandler extends DefaultMessageHandler {
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactoryTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactoryTest.java
new file mode 100644
index 00000000000..af07798fddc
--- /dev/null
+++
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactoryTest.java
@@ -0,0 +1,188 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.starter.helix;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.messages.TableDeletionMessage;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.core.data.manager.InstanceDataManager;
+import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.ComplexFieldSpec;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.mockito.InOrder;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Covers the OPEN_STRUCT gauge keys resolved on table deletion.
`OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT` is a
+/// keyed gauge, so the generic `removeTableGauge(table, gauge)` sweep cannot
reach it; the handler
+/// re-derives its keys from the table config instead.
+public class SegmentMessageHandlerFactoryTest {
+
+ private static Schema schema() {
+ return new Schema.SchemaBuilder().setSchemaName("t")
+ .addField(new ComplexFieldSpec("metrics", DataType.OPEN_STRUCT, true,
Map.of()))
+ .addField(new DimensionFieldSpec("plain", DataType.STRING, true))
+ .build();
+ }
+
+ private static TableConfig tableConfig(List<FieldConfig> fieldConfigs) {
+ return new
TableConfigBuilder(TableType.OFFLINE).setTableName("t").setFieldConfigList(fieldConfigs).build();
+ }
+
+ private static FieldConfig openStructField(String column, String
denseKeysJson)
+ throws Exception {
+ JsonNode indexes = JsonUtils.stringToJsonNode("{\"open_struct\":
{\"denseKeys\": " + denseKeysJson + "}}");
+ return new FieldConfig.Builder(column).withIndexes(indexes).build();
+ }
+
+ @Test
+ public void testResolvesOneMetricKeyPerConfiguredDenseKey()
+ throws Exception {
+ TableConfig tableConfig = tableConfig(List.of(openStructField("metrics",
"[\"clicks\", \"views\"]")));
+
+ List<String> metricKeys =
SegmentMessageHandlerFactory.openStructMetricKeys(tableConfig, schema());
+
+ assertEquals(metricKeys.size(), 2);
+ assertTrue(metricKeys.contains("metrics$clicks"), metricKeys.toString());
+ assertTrue(metricKeys.contains("metrics$views"), metricKeys.toString());
+ }
+
+ /// The emitted gauge key is escaped for JMX, so the key used to remove it
has to be escaped the
+ /// same way or the removal silently misses.
+ @Test
+ public void testMetricKeysAreEscapedLikeTheEmittedGauge()
+ throws Exception {
+ TableConfig tableConfig = tableConfig(List.of(openStructField("metrics",
"[\"promo\\\"code\"]")));
+
+ List<String> metricKeys =
SegmentMessageHandlerFactory.openStructMetricKeys(tableConfig, schema());
+
+ assertEquals(metricKeys, List.of("metrics$promo%22code"));
+ }
+
+ @Test
+ public void testNoKeysWithoutConfiguredDenseKeys()
+ throws Exception {
+ // No configured denseKeys → nothing recoverable for removal. When
perKeyMetricsEnabled is on,
+ // discovered-key gauges are emitted but not reachable here; they survive
until server restart.
+ TableConfig tableConfig = tableConfig(List.of(openStructField("metrics",
"[]")));
+
+ assertTrue(SegmentMessageHandlerFactory.openStructMetricKeys(tableConfig,
schema()).isEmpty());
+ }
+
+ @Test
+ public void testNoKeysForTableWithoutOpenStructColumns() {
+ Schema plainSchema = new Schema.SchemaBuilder().setSchemaName("t")
+ .addField(new DimensionFieldSpec("plain", DataType.STRING, true))
+ .build();
+
+
assertTrue(SegmentMessageHandlerFactory.openStructMetricKeys(tableConfig(List.of()),
plainSchema).isEmpty());
+ }
+
+ private static final String TABLE = "t_OFFLINE";
+
+ private static void runDeletion(InstanceDataManager instanceDataManager,
ServerMetrics serverMetrics)
+ throws Exception {
+ new SegmentMessageHandlerFactory(instanceDataManager,
serverMetrics).createHandler(
+ new TableDeletionMessage(TABLE), null).handleMessage();
+ }
+
+ private static InstanceDataManager
instanceDataManagerReturning(Pair<TableConfig, Schema> cachedConfigAndSchema) {
+ TableDataManager tableDataManager = mock(TableDataManager.class);
+
when(tableDataManager.getCachedTableConfigAndSchema()).thenReturn(cachedConfigAndSchema);
+ InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
+
when(instanceDataManager.getTableDataManager(TABLE)).thenReturn(tableDataManager);
+ return instanceDataManager;
+ }
+
+ /// The reason this change exists: the generic sweep in the handler composes
only the unkeyed
+ /// `<gauge>.<table>`, so without this targeted call the per-key gauge
outlives the table. Asserts on
+ /// the keyed `removeTableGauge` overload rather than on a registry, because
reading back a gauge
+ /// written by `setOrUpdateTableGauge` needs a metrics factory that
pinot-server does not register.
+ @Test
+ public void testHandlerRemovesTheEmittedPerKeyGauge()
+ throws Exception {
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ TableConfig tableConfig = tableConfig(List.of(openStructField("metrics",
"[\"clicks\"]")));
+
+ runDeletion(instanceDataManagerReturning(Pair.of(tableConfig, schema())),
serverMetrics);
+
+ verify(serverMetrics).removeTableGauge(TABLE, "metrics$clicks",
ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT);
+ }
+
+ /// The keys have to be read off the table data manager before deleteTable
discards it.
+ @Test
+ public void testKeysAreResolvedBeforeTheTableIsDeleted()
+ throws Exception {
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ TableConfig tableConfig = tableConfig(List.of(openStructField("metrics",
"[\"clicks\"]")));
+ InstanceDataManager instanceDataManager =
instanceDataManagerReturning(Pair.of(tableConfig, schema()));
+
+ runDeletion(instanceDataManager, serverMetrics);
+
+ InOrder inOrder = inOrder(instanceDataManager, serverMetrics);
+ inOrder.verify(instanceDataManager).getTableDataManager(TABLE);
+ inOrder.verify(instanceDataManager).deleteTable(eq(TABLE), anyLong());
+ inOrder.verify(serverMetrics)
+ .removeTableGauge(TABLE, "metrics$clicks",
ServerGauge.OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT);
+ }
+
+ /// Metric bookkeeping must never block table deletion.
+ @Test
+ public void testDeletionProceedsWhenKeysCannotBeResolved()
+ throws Exception {
+ InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
+ when(instanceDataManager.getTableDataManager(TABLE)).thenThrow(new
IllegalStateException("boom"));
+
+ runDeletion(instanceDataManager, mock(ServerMetrics.class));
+
+ verify(instanceDataManager, times(1)).deleteTable(eq(TABLE), anyLong());
+ }
+
+ @Test
+ public void testDeletionProceedsWithoutATableDataManager()
+ throws Exception {
+ InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
+ when(instanceDataManager.getTableDataManager(TABLE)).thenReturn(null);
+
+ runDeletion(instanceDataManager, mock(ServerMetrics.class));
+
+ verify(instanceDataManager, times(1)).deleteTable(eq(TABLE), anyLong());
+ }
+}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
index ad141bc7276..0c09fde132c 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
@@ -62,6 +62,7 @@ public class OpenStructIndexConfig extends IndexConfig {
private final double _denseKeyMinFillRate;
private final List<FieldConfig> _valueFieldConfigs;
private final boolean _sparseJsonIndex;
+ private final boolean _perKeyMetricsEnabled;
// Eager lookup from key name → FieldConfig for O(1) per-key access. Built
in constructor
// so the config is fully immutable and safe to share across threads.
private final Map<String, FieldConfig> _valueFieldConfigIndex;
@@ -79,6 +80,16 @@ public class OpenStructIndexConfig extends IndexConfig {
this(disabled, defaultValueFieldConfig, maxDenseKeys, denseKeys,
denseKeyMinFillRate, valueFieldConfigs, null);
}
+ /// @deprecated Use the 8-arg constructor accepting `perKeyMetricsEnabled`.
Kept for binary
+ /// compatibility with existing callers built against the
pre-`perKeyMetricsEnabled` signature.
+ @Deprecated
+ public OpenStructIndexConfig(Boolean disabled, @Nullable FieldConfig
defaultValueFieldConfig,
+ @Nullable Integer maxDenseKeys, @Nullable Set<String> denseKeys,
@Nullable Double denseKeyMinFillRate,
+ @Nullable List<FieldConfig> valueFieldConfigs, @Nullable Boolean
sparseJsonIndex) {
+ this(disabled, defaultValueFieldConfig, maxDenseKeys, denseKeys,
denseKeyMinFillRate, valueFieldConfigs,
+ sparseJsonIndex, null);
+ }
+
@JsonCreator
public OpenStructIndexConfig(
@JsonProperty("disabled") Boolean disabled,
@@ -87,7 +98,8 @@ public class OpenStructIndexConfig extends IndexConfig {
@JsonProperty("denseKeys") @Nullable Set<String> denseKeys,
@JsonProperty("denseKeyMinFillRate") @Nullable Double
denseKeyMinFillRate,
@JsonProperty("valueFieldConfigs") @Nullable List<FieldConfig>
valueFieldConfigs,
- @JsonProperty("sparseJsonIndex") @Nullable Boolean sparseJsonIndex) {
+ @JsonProperty("sparseJsonIndex") @Nullable Boolean sparseJsonIndex,
+ @JsonProperty("perKeyMetricsEnabled") @Nullable Boolean
perKeyMetricsEnabled) {
super(disabled);
_defaultValueFieldConfig = defaultValueFieldConfig;
_maxDenseKeys = maxDenseKeys != null ? maxDenseKeys :
DEFAULT_MAX_DENSE_KEYS;
@@ -95,6 +107,7 @@ public class OpenStructIndexConfig extends IndexConfig {
_denseKeyMinFillRate = denseKeyMinFillRate != null ? denseKeyMinFillRate :
DEFAULT_DENSE_KEY_MIN_FILL_RATE;
_valueFieldConfigs = valueFieldConfigs;
_sparseJsonIndex = sparseJsonIndex != null && sparseJsonIndex;
+ _perKeyMetricsEnabled = perKeyMetricsEnabled != null &&
perKeyMetricsEnabled;
if (valueFieldConfigs == null || valueFieldConfigs.isEmpty()) {
_valueFieldConfigIndex = Map.of();
} else {
@@ -181,6 +194,15 @@ public class OpenStructIndexConfig extends IndexConfig {
return _sparseJsonIndex;
}
+ /// When `true`, `OPEN_STRUCT_LAST_SEGMENT_KEY_DOC_COUNT` is emitted for
every key present in the
+ /// sealed segment — dense, sparse, configured, or discovered. The cost is
that the number of
+ /// metrics-registry entries follows the ingested key space, and table
deletion can only sweep keys
+ /// recoverable from `denseKeys`. When `false` (default), the gauge fires
only for keys named in
+ /// `denseKeys`.
+ public boolean isPerKeyMetricsEnabled() {
+ return _perKeyMetricsEnabled;
+ }
+
private static boolean invertedFromIndexes(FieldConfig fieldConfig, String
key) {
JsonNode indexes = fieldConfig.getIndexes();
if (indexes == null || !indexes.isObject()) {
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/data/OpenStructNaming.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/data/OpenStructNaming.java
index a0061c6d1a4..a64b25d0443 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/data/OpenStructNaming.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/data/OpenStructNaming.java
@@ -22,6 +22,9 @@ package org.apache.pinot.spi.data;
/// Naming convention for OPEN_STRUCT materialized columns. Each dense
OPEN_STRUCT key is stored as
/// a column named `<openStructColumn>$<key>`. Sparse keys share a single
synthetic JSON column
/// named `<openStructColumn>$__sparse__`.
+///
+/// [#metricKey] builds a superficially similar `<openStructColumn>$<key>`
string for per-key metric
+/// names, but escapes the key for JMX export and is not a column name -- see
its docs.
public final class OpenStructNaming {
public static final String SEPARATOR = "$";
public static final String SPARSE_SUFFIX = "__sparse__";
@@ -33,6 +36,23 @@ public final class OpenStructNaming {
return openStructColumn + SEPARATOR + key;
}
+ /// Builds the `<openStructColumn>$<key>` identifier used as the key segment
of a per-key OPEN_STRUCT
+ /// metric name; the scrape rule splits it back into `column` and `key`
labels. Not a column name --
+ /// the key need not have one on disk, and the escaped output must not be
fed to [#parseKey] or
+ /// [#parseParentColumn].
+ ///
+ /// Percent-escapes the four characters `javax.management.ObjectName.quote`
backslash-escapes on the
+ /// way to JMX (`"`, `\`, `*`, `?`), plus `%` itself so the mapping stays
reversible. An escaped `"`
+ /// stops the name matching the per-key rule in
+ /// `docker/images/pinot/etc/jmx_prometheus_javaagent/configs/server.yml` at
all; the other three
+ /// leave a stray backslash in the label value. Nothing else is touched, so
`user.id`, `user-id` and
+ /// `user_id` stay distinct series.
+ public static String metricKey(String openStructColumn, String key) {
+ // '%' first so the escapes introduced below are not themselves re-escaped.
+ return openStructColumn + SEPARATOR + key.replace("%",
"%25").replace("\"", "%22").replace("\\", "%5C")
+ .replace("*", "%2A").replace("?", "%3F");
+ }
+
public static String sparseColumnName(String openStructColumn) {
return openStructColumn + SEPARATOR + SPARSE_SUFFIX;
}
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/config/table/OpenStructIndexConfigTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/config/table/OpenStructIndexConfigTest.java
index 48e58935fc0..d4f26c23702 100644
---
a/pinot-spi/src/test/java/org/apache/pinot/spi/config/table/OpenStructIndexConfigTest.java
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/config/table/OpenStructIndexConfigTest.java
@@ -46,6 +46,28 @@ public class OpenStructIndexConfigTest {
assertTrue(config.shouldUseDictionaryForKey("any"));
}
+ @Test
+ public void testPerKeyMetricsDefaultsToFalse() {
+ // 7-arg constructor (pre-existing callers) — must default to false.
+ OpenStructIndexConfig config = new OpenStructIndexConfig(false, null, -1,
null, 0.5, null, null);
+ assertFalse(config.isPerKeyMetricsEnabled());
+ // DEFAULT constant.
+ assertFalse(OpenStructIndexConfig.DEFAULT.isPerKeyMetricsEnabled());
+ }
+
+ @Test
+ public void testPerKeyMetricsRoundTripsFromJson()
+ throws Exception {
+ String json = "{\"perKeyMetricsEnabled\": true, \"denseKeys\":
[\"clicks\"]}";
+ OpenStructIndexConfig config = JsonUtils.stringToObject(json,
OpenStructIndexConfig.class);
+ assertTrue(config.isPerKeyMetricsEnabled());
+ assertEquals(config.getDenseKeys(), Set.of("clicks"));
+
+ // Absent field → false.
+ OpenStructIndexConfig noFlag = JsonUtils.stringToObject("{}",
OpenStructIndexConfig.class);
+ assertFalse(noFlag.isPerKeyMetricsEnabled());
+ }
+
@Test
public void testDisabledConfig() {
OpenStructIndexConfig config = OpenStructIndexConfig.DISABLED;
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/OpenStructNamingTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/OpenStructNamingTest.java
index 5c057d4acac..db7c5c8e38b 100644
---
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/OpenStructNamingTest.java
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/OpenStructNamingTest.java
@@ -18,10 +18,14 @@
*/
package org.apache.pinot.spi.data;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;
@@ -72,4 +76,64 @@ public class OpenStructNamingTest {
public void testParseKeyRejectsNonMaterialized() {
OpenStructNaming.parseKey("metrics");
}
+
+ /// Only the four characters ObjectName.quote backslash-escapes are escaped,
plus '%' itself so the
+ /// mapping stays reversible. Everything else has to survive, or keys that
differ only in punctuation
+ /// would collapse into one metric series.
+ @Test
+ public void testMetricKeyEscapesOnlyObjectNameEscapedChars() {
+ assertEquals(OpenStructNaming.metricKey("metrics", "pro\"mo"),
"metrics$pro%22mo");
+ assertEquals(OpenStructNaming.metricKey("metrics", "pro\\mo"),
"metrics$pro%5Cmo");
+ assertEquals(OpenStructNaming.metricKey("metrics", "pro*mo"),
"metrics$pro%2Amo");
+ assertEquals(OpenStructNaming.metricKey("metrics", "pro?mo"),
"metrics$pro%3Fmo");
+ assertEquals(OpenStructNaming.metricKey("metrics", "pro%mo"),
"metrics$pro%25mo");
+
+ // Untouched: these are all safe inside a quoted ObjectName and in a
Prometheus label value.
+ assertEquals(OpenStructNaming.metricKey("metrics",
"clicks.v2$promo-code"), "metrics$clicks.v2$promo-code");
+ assertEquals(OpenStructNaming.metricKey("metrics", "a,b"), "metrics$a,b");
+ assertEquals(OpenStructNaming.metricKey("metrics", "a=b"), "metrics$a=b");
+ assertEquals(OpenStructNaming.metricKey("metrics", "a b"), "metrics$a b");
+ assertEquals(OpenStructNaming.metricKey("metrics", "user_id"),
"metrics$user_id");
+ }
+
+ /// The escaping must be injective. Folding the four to '_' would not be:
'_' is itself a legal key
+ /// character, so 'a"b' and 'a_b' would land on the same gauge and each seal
would silently overwrite
+ /// the other. '%' is escaped for the same reason -- without it 'a%22b'
would collide with 'a"b'.
+ @Test
+ public void testMetricKeyEscapingIsInjective() {
+ assertNotEquals(OpenStructNaming.metricKey("metrics", "a\"b"),
OpenStructNaming.metricKey("metrics", "a_b"));
+ assertNotEquals(OpenStructNaming.metricKey("metrics", "a\"b"),
OpenStructNaming.metricKey("metrics", "a%22b"));
+ assertNotEquals(OpenStructNaming.metricKey("metrics", "a\\b"),
OpenStructNaming.metricKey("metrics", "a%5Cb"));
+
+ // The four escaped characters must not collide with each other either.
+ Set<String> encoded = new HashSet<>();
+ for (String key : List.of("a\"b", "a\\b", "a*b", "a?b", "a%b", "a_b")) {
+ assertTrue(encoded.add(OpenStructNaming.metricKey("metrics", key)),
"collision on key: " + key);
+ }
+ }
+
+ /// user.id / user-id / user_id must stay distinct series -- the reason
escaping is narrow.
+ @Test
+ public void testMetricKeyKeepsPunctuationVariantsDistinct() {
+ String a = OpenStructNaming.metricKey("metrics", "user.id");
+ String b = OpenStructNaming.metricKey("metrics", "user-id");
+ String c = OpenStructNaming.metricKey("metrics", "user_id");
+ assertNotEquals(a, b);
+ assertNotEquals(b, c);
+ assertNotEquals(a, c);
+ }
+
+ /// A key needing no escaping must produce exactly the materialized column
name, so the dense-key
+ /// metric and its on-disk column stay addressable by the same string in the
common case.
+ @Test
+ public void testMetricKeyMatchesMaterializedNameWhenNoEscapingNeeded() {
+ assertEquals(OpenStructNaming.metricKey("metrics", "clicks"),
+ OpenStructNaming.materializedColumnName("metrics", "clicks"));
+ }
+
+ @Test
+ public void testMetricKeyHandlesEmptyAndAllEscapedKeys() {
+ assertEquals(OpenStructNaming.metricKey("metrics", ""), "metrics$");
+ assertEquals(OpenStructNaming.metricKey("metrics", "\"\\*?%"),
"metrics$%22%5C%2A%3F%25");
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
