Caideyipi commented on code in PR #17936:
URL: https://github.com/apache/iotdb/pull/17936#discussion_r3464926823
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java:
##########
@@ -89,12 +106,57 @@ public void removeUserData() {
resultSets.forEach(SubscriptionResultSet::removeUserData);
}
+ private static boolean resolveTimeSelected(
+ final Map<String, Map<String, Boolean>> timeSelectedByTable,
+ final boolean defaultTimeSelected,
+ final String databaseName,
+ final Tablet tablet) {
+ if (Objects.isNull(timeSelectedByTable) || timeSelectedByTable.isEmpty()) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> tableMap =
timeSelectedByTable.get(databaseName);
+ if (Objects.isNull(tablet)) {
+ return defaultTimeSelected;
+ }
Review Comment:
Fixed. `SubscriptionRecordHandler` now validates `tablet`, `databaseName`,
and `tablet.getTableName()` before reading from the map.
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java:
##########
@@ -89,12 +106,57 @@ public void removeUserData() {
resultSets.forEach(SubscriptionResultSet::removeUserData);
}
+ private static boolean resolveTimeSelected(
+ final Map<String, Map<String, Boolean>> timeSelectedByTable,
+ final boolean defaultTimeSelected,
+ final String databaseName,
+ final Tablet tablet) {
+ if (Objects.isNull(timeSelectedByTable) || timeSelectedByTable.isEmpty()) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> tableMap =
timeSelectedByTable.get(databaseName);
+ if (Objects.isNull(tablet)) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> resolvedTableMap =
+ Objects.nonNull(tableMap)
+ ? tableMap
+ : getIgnoreCase(timeSelectedByTable, databaseName, null);
Review Comment:
Fixed. `timeSelectedByTable` db/table keys are normalized to lowercase when
copied, and the handler now performs direct lowercase lookups instead of
scanning case-insensitively.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java:
##########
@@ -374,41 +372,41 @@ private void validateConsensusProtocolSupport(final
TopicConfig topicConfig)
throw new SubscriptionException(exceptionMessage);
}
- private void validateConsensusTableColumnPattern(final TopicConfig
topicConfig)
- throws SubscriptionException {
- if (!topicConfig.hasAttribute(TopicConstant.COLUMN_KEY)) {
- return;
+ private void validateColumnFilter(final TopicConfig topicConfig) throws
SubscriptionException {
+ int columnFilterKeyCount = 0;
+ for (final String key : topicConfig.getAttribute().keySet()) {
+ if (TopicConstant.COLUMN_FILTER_KEY.equalsIgnoreCase(key)) {
+ columnFilterKeyCount++;
+ }
}
-
- if (!topicConfig.isTableTopic()) {
+ if (columnFilterKeyCount > 1) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, %s is only supported for table
topics",
- TopicConstant.COLUMN_KEY);
+ "Failed to create or alter topic, duplicate %s attributes are
not allowed",
+ TopicConstant.COLUMN_FILTER_KEY);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
Review Comment:
Fixed for consistency. Known topic config keys are now checked
case-insensitively for duplicates, not only `column-filter`; added a `mode`
duplicate-key validation test as coverage.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java:
##########
@@ -313,6 +313,28 @@ public long getInstanceVersion() {
return instanceVersion.get();
}
+ public Map<String, Map<String, TsTable>> getTableSnapshot() {
+ readWriteLock.readLock().lock();
+ try {
+ return databaseTableMap.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
+ entry.getValue().entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ tableEntry -> new
TsTable(tableEntry.getValue()),
+ (left, right) -> right,
+ ConcurrentHashMap::new)),
+ (left, right) -> right,
+ ConcurrentHashMap::new));
+ } finally {
Review Comment:
Fixed. The snapshot collector now uses `HashMap` instead of
`ConcurrentHashMap`.
--
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]
