raghavyadav01 commented on code in PR #19041:
URL: https://github.com/apache/pinot/pull/19041#discussion_r3752369533
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -246,17 +254,57 @@ public void seal()
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 =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercionFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures (keys: {})",
+ _columnName, totalCoercionFailures, _coercionFailuresPerKey);
+ }
+ long totalInferenceFailures =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInferenceFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': {} type inference failures fell back to
STRING (keys: {})",
+ _columnName, totalInferenceFailures, _inferenceFailuresPerKey);
}
+ emitMetrics(sparseKeys.size());
emitParentColumnMetadata(sparseKeys);
}
+ private void emitMetrics(int sparseKeyCount) {
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics == null || _numDocs == 0) {
+ return;
+ }
+ String col = _columnName;
+
+ long totalCoercion =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercion > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, totalCoercion);
+ }
+ long totalInference =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInference > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, totalInference);
+ }
+
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_DENSE_KEY_COUNT, _resolvedDenseKeys.size());
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_SPARSE_KEY_COUNT, sparseKeyCount);
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_TOTAL_KEYS_DISCOVERED,
_presenceBitmaps.size());
+
+ for (String key : _resolvedDenseKeys) {
+ RoaringBitmap presence = _presenceBitmaps.get(key);
+ if (presence != null) {
+ long fillPct = (long) presence.getCardinality() * 100 / _numDocs;
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType,
+ OpenStructNaming.materializedColumnName(col, key),
Review Comment:
Have you checked how this one scrapes? `OpenStructNaming.SEPARATOR` is `$`,
and every key/partition rule in `server.yml` uses `\w+` for that segment, so I
don't think `openStructKeyFillRate.<table>.metrics$clicks` matches any rule —
it would fall through to the javaagent default name with the key baked into the
metric name and no `table`/`tableType` labels. Same for the per-key meters in
`MutableOpenStructIndex`.
Would it make sense to either drop the per-key dimension here, or add a
`server.yml` rule that exports the key as a `column` label (with `$` sanitized)?
On verification — `ServerPrometheusMetricsTest` is data-provider-driven over
all `ServerGauge` values, so these do get exercised, but only with plain table
names and int partitions. Could we add a case with a `$`-bearing key so CI
actually covers the exported shape?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java:
##########
@@ -162,7 +171,9 @@ private Object tryCoerce(String key, Object rawValue,
DataType storedType) {
} catch (Exception e) {
ServerMetrics serverMetrics = ServerMetrics.get();
if (serverMetrics != null) {
-
serverMetrics.addMeteredGlobalValue(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
1);
+ serverMetrics.addMeteredTableValue(_tableNameWithType,
+ OpenStructNaming.materializedColumnName(_openStructColumn, key),
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1);
Review Comment:
Does the unbounded-registry concern from the earlier thread still apply on
this path? The splitter side was moved to `_resolvedDenseKeys`, but both meters
here are still keyed on the raw ingested key, and meters are never removed from
the registry — so a stream whose map keys carry ids would mint one meter per
key for the lifetime of the process.
Also, on the "config-bounded" reasoning for the splitter:
`OpenStructIndexConfig.DEFAULT_MAX_DENSE_KEYS` is `-1` and `denseKeys` defaults
to empty, so with default config `_resolvedDenseKeys` is every key at >= 50%
fill — data-driven, not config-driven. Across segments the union of those keys
is unbounded too, and each one leaves a gauge behind holding whatever the
last-sealed segment wrote.
Would you be open to keeping both meters at column granularity (the per-key
detail is already in the log line), and gating the fill-rate gauge on an
explicit `denseKeys` list or `maxDenseKeys > 0`?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -246,17 +254,57 @@ public void seal()
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 =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercionFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures (keys: {})",
+ _columnName, totalCoercionFailures, _coercionFailuresPerKey);
+ }
+ long totalInferenceFailures =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInferenceFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': {} type inference failures fell back to
STRING (keys: {})",
+ _columnName, totalInferenceFailures, _inferenceFailuresPerKey);
}
+ emitMetrics(sparseKeys.size());
emitParentColumnMetadata(sparseKeys);
}
+ private void emitMetrics(int sparseKeyCount) {
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics == null || _numDocs == 0) {
+ return;
+ }
+ String col = _columnName;
+
+ long totalCoercion =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercion > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, totalCoercion);
+ }
+ long totalInference =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInference > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, totalInference);
+ }
+
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_DENSE_KEY_COUNT, _resolvedDenseKeys.size());
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_SPARSE_KEY_COUNT, sparseKeyCount);
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_TOTAL_KEYS_DISCOVERED,
_presenceBitmaps.size());
Review Comment:
What's the intended reading of these three when a server hosts many segments
of the same table? They're keyed on `(table, column)` with no segment
dimension, so each seal overwrites the previous one and the value ends up
meaning "whatever the last segment to seal here saw".
`OPEN_STRUCT_TOTAL_KEYS_DISCOVERED` in particular reads like a table-wide count.
Also, nothing calls `removeTableGauge` — is it worth clearing these when the
table is deleted, or is the stale-value-until-restart behaviour acceptable here?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -201,7 +205,11 @@ private void addMap(@Nullable Map<String, Object> map) {
? keySpec.getDataType()
: _inferredTypes.computeIfAbsent(key, k -> {
DataType inferred =
OpenStructTypeInference.inferDataType(rawValue);
- return inferred != null ? inferred : DataType.STRING;
+ if (inferred == null) {
+ _inferenceFailuresPerKey.merge(key, 1L, Long::sum);
+ return DataType.STRING;
+ }
+ return inferred;
});
Review Comment:
Should this increment live outside `computeIfAbsent`? The mapping function
only runs on the first sighting of a key, so a million unmappable values on one
key records `1` — but the meter's unit is `"values"` and the description says
"Number of OPEN_STRUCT entries dropped".
Related: nothing is dropped here, the key falls back to STRING.
`MutableOpenStructIndex.resolveStoredType` hits the same condition and returns
`null`, dropping the entry. Is the same meter meant to cover both? A Map- or
List-valued key would then be absent in the consuming segment but materialized
as `toString()` output after seal, which `OpenStructConsumingSealedParityTest`
doesn't currently cover — worth a case there if that divergence is intentional.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -246,17 +254,57 @@ public void seal()
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 =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercionFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures (keys: {})",
+ _columnName, totalCoercionFailures, _coercionFailuresPerKey);
+ }
+ long totalInferenceFailures =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInferenceFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': {} type inference failures fell back to
STRING (keys: {})",
+ _columnName, totalInferenceFailures, _inferenceFailuresPerKey);
}
+ emitMetrics(sparseKeys.size());
emitParentColumnMetadata(sparseKeys);
}
+ private void emitMetrics(int sparseKeyCount) {
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics == null || _numDocs == 0) {
+ return;
+ }
+ String col = _columnName;
+
+ long totalCoercion =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercion > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, totalCoercion);
+ }
+ long totalInference =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInference > 0) {
+ serverMetrics.addMeteredTableValue(_tableNameWithType, col,
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, totalInference);
+ }
+
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_DENSE_KEY_COUNT, _resolvedDenseKeys.size());
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_SPARSE_KEY_COUNT, sparseKeyCount);
+ serverMetrics.setOrUpdateTableGauge(_tableNameWithType, col,
+ ServerGauge.OPEN_STRUCT_TOTAL_KEYS_DISCOVERED,
_presenceBitmaps.size());
+
+ for (String key : _resolvedDenseKeys) {
+ RoaringBitmap presence = _presenceBitmaps.get(key);
+ if (presence != null) {
+ long fillPct = (long) presence.getCardinality() * 100 / _numDocs;
Review Comment:
Is integer truncation OK here? `classify()` adds configured `denseKeys`
before the fill-rate check, so a key present in 3 of 100k docs is dense and
reports `0` — indistinguishable from no data, though that's exactly the case
worth flagging (a configured dense key wasting a materialized column). Would
scaling by 10000, or emitting cardinality and `_numDocs` separately, work
better?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -246,17 +254,57 @@ public void seal()
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 =
_coercionFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalCoercionFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': dropped {} values due to type coercion
failures (keys: {})",
+ _columnName, totalCoercionFailures, _coercionFailuresPerKey);
+ }
+ long totalInferenceFailures =
_inferenceFailuresPerKey.values().stream().mapToLong(Long::longValue).sum();
+ if (totalInferenceFailures > 0) {
+ LOGGER.info("OPEN_STRUCT '{}': {} type inference failures fell back to
STRING (keys: {})",
+ _columnName, totalInferenceFailures, _inferenceFailuresPerKey);
}
Review Comment:
Should these be DEBUG? The key space is user-controlled, so printing the
whole map at INFO on every seal could flood the logs for a wide struct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
