Caideyipi commented on code in PR #17936:
URL: https://github.com/apache/iotdb/pull/17936#discussion_r3464927663
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java:
##########
@@ -93,14 +94,62 @@ public TPushTopicMetaRespExceptionMessage
handleSingleTopicMetaChanges(
private void handleSingleTopicMetaChangesInternal(final TopicMeta
metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
- TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(topicName), metaFromCoordinator);
+ final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName);
+ TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
topicMetaKeeper.removeTopicMeta(topicName);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
+ if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
+ SubscriptionAgent.broker().refreshColumnFilter(topicName,
metaFromCoordinator.getConfig());
+ } else if (!metaFromCoordinator.getConfig().isTableTopic()) {
+ SubscriptionAgent.broker().dropColumnFilter(topicName);
+ }
Review Comment:
ConfigNode validation rejects create/alter of tree topics with
`column-filter`. The DataNode-side drop is kept defensive for
stale/replayed/hot-updated metadata so a tree topic cannot retain an old
column-filter matcher.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java:
##########
@@ -93,14 +94,62 @@ public TPushTopicMetaRespExceptionMessage
handleSingleTopicMetaChanges(
private void handleSingleTopicMetaChangesInternal(final TopicMeta
metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
- TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(topicName), metaFromCoordinator);
+ final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName);
+ TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
topicMetaKeeper.removeTopicMeta(topicName);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
+ if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
+ SubscriptionAgent.broker().refreshColumnFilter(topicName,
metaFromCoordinator.getConfig());
+ } else if (!metaFromCoordinator.getConfig().isTableTopic()) {
+ SubscriptionAgent.broker().dropColumnFilter(topicName);
+ }
SubscriptionAgent.broker()
.refreshConsensusQueueOrderMode(topicName,
metaFromCoordinator.getConfig().getOrderMode());
}
+ static boolean shouldRefreshColumnFilter(final TopicMeta oldMeta, final
TopicMeta newMeta) {
+ if (Objects.isNull(newMeta) || !newMeta.getConfig().isTableTopic()) {
+ return false;
+ }
+ if (Objects.isNull(oldMeta) || !oldMeta.getConfig().isTableTopic()) {
+ return true;
+ }
+
+ final TopicConfig oldConfig = oldMeta.getConfig();
+ final TopicConfig newConfig = newMeta.getConfig();
+ return !Objects.equals(
+ normalizeColumnFilterBindingValue(oldConfig.getColumnFilter()),
+ normalizeColumnFilterBindingValue(newConfig.getColumnFilter()))
+ || !Objects.equals(
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE)),
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ newConfig, TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE)))
+ || !Objects.equals(
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.TABLE_KEY,
TopicConstant.TABLE_DEFAULT_VALUE)),
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ newConfig, TopicConstant.TABLE_KEY,
TopicConstant.TABLE_DEFAULT_VALUE)));
+ }
+
+ private static String getAttributeIgnoreCase(
+ final TopicConfig topicConfig, final String key, final String
defaultValue) {
+ return topicConfig.getAttribute().entrySet().stream()
+ .filter(entry -> key.equalsIgnoreCase(entry.getKey()))
+ .map(Map.Entry::getValue)
+ .filter(Objects::nonNull)
+ .findFirst()
+ .orElse(defaultValue);
+ }
Review Comment:
The hot-refresh comparison normalizes the relevant values before comparing.
I kept the original topic attributes unchanged here because
`TopicConfig`/`PipeParameters` own attribute lookup semantics, and mutating
pushed coordinator metadata in the handler would be broader than needed.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java:
##########
@@ -458,43 +472,18 @@ private List<Tablet> convertRelationalInsertRowNode(final
RelationalInsertRowNod
final String[] measurements = node.getMeasurements();
final TSDataType[] dataTypes = node.getDataTypes();
final Object[] values = node.getValues();
- final List<Integer> matchedColumnIndices =
- getMatchedTableColumnIndices(
- measurements, dataTypes, values, node.getColumnCategories(),
false);
- if (matchedColumnIndices.isEmpty()) {
- return Collections.emptyList();
- }
-
- final int columnCount = matchedColumnIndices.size();
- final List<String> columnNames = new ArrayList<>(columnCount);
- final List<TSDataType> columnDataTypes = new ArrayList<>(columnCount);
- final List<ColumnCategory> columnTypes = new ArrayList<>(columnCount);
- for (final int originalColIdx : matchedColumnIndices) {
- columnNames.add(measurements[originalColIdx]);
- columnDataTypes.add(dataTypes[originalColIdx]);
- columnTypes.add(toTsFileColumnCategory(node.getColumnCategories(),
originalColIdx));
- }
-
final Tablet tablet =
- new Tablet(
- tableName != null ? tableName : "", columnNames, columnDataTypes,
columnTypes, 1);
- tablet.addTimestamp(0, time);
-
- for (int i = 0; i < columnCount; i++) {
- final int originalColIdx = matchedColumnIndices.get(i);
- final Object value = values[originalColIdx];
- if (value == null) {
- if (tablet.getBitMaps() == null) {
- tablet.initBitMaps();
- }
- tablet.getBitMaps()[i].mark(0);
- } else {
- addValueToTablet(tablet, 0, i, dataTypes[originalColIdx], value);
- }
+ buildTableModelTabletFromRow(
+ tableName, time, measurements, dataTypes, values,
node.getColumnCategories());
+ if (Objects.isNull(tablet)) {
+ return Collections.emptyList();
}
- tablet.setRowSize(1);
- return Collections.singletonList(tablet);
+ final Tablet prunedTablet =
+ TabletColumnPruner.pruneTableModelTablet(tablet, databaseName,
getColumnFilterMatcher());
Review Comment:
Partially folded into construction for tablet nodes: malformed columns are
filtered while building the tablet arrays so schemas/columns/bitmaps stay
aligned. The actual column-filter pruning remains in `TabletColumnPruner` so
row/tablet paths share the same tag/attribute/time-only behavior.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java:
##########
@@ -1813,6 +1819,25 @@ private boolean createAndEnqueueEvent(
return true;
}
+ private Map<String, Map<String, Boolean>> getTimeSelectedByTable(
+ final String databaseName, final List<Tablet> tablets) {
+ if (Objects.isNull(databaseName) || Objects.isNull(tablets) ||
tablets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ final ColumnFilterMatcher matcher =
+ SubscriptionAgent.broker().getColumnFilterMatcher(topicName);
+ final Map<String, Boolean> tableMap = new HashMap<>();
+ for (final Tablet tablet : tablets) {
+ if (Objects.nonNull(tablet) && Objects.nonNull(tablet.getTableName())) {
+ tableMap.put(
+ tablet.getTableName(), matcher.isTimeSelected(databaseName,
tablet.getTableName()));
+ }
+ }
+ return tableMap.isEmpty()
+ ? Collections.emptyMap()
+ : Collections.singletonMap(databaseName, tableMap);
+ }
+
Review Comment:
I kept the broker lookup here intentionally so the queue observes the latest
matcher after topic-meta hot refresh. Passing a matcher captured earlier could
become stale while the queue keeps running.
--
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]
