JackieTien97 commented on code in PR #12299:
URL: https://github.com/apache/iotdb/pull/12299#discussion_r1555134830
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
+ return path.getDevicePath().matchFullPath(keyPath);
+ } catch (IllegalPathException e) {
+ return false;
+ }
+ };
+ }
+ if (PathPatternUtil.isMultiLevelMatchWildcard(measurement)) {
+ measurementFilter = m -> true;
+ }
+ if (deviceFilter == null) {
+ deviceFilter = d ->
d.toString().equals(path.getDevicePath().getFullPath());
+ }
+
+ if (measurementFilter == null) {
+ measurementFilter = m -> PathPatternUtil.isNodeMatch(measurement,
m.toString());
+ }
+
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (Boolean.TRUE.equals(deviceFilter.apply(device))) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ if (Boolean.TRUE.equals(measurementFilter.apply(entry.getKey()))) {
+ synchronized (entry) {
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+
cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void invalidateDataRegionLastCache(String database) {
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (device.toString().startsWith(database)) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+ cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
Review Comment:
why don't use`synchronized (entry) ` here?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
Review Comment:
`path.getDevicePath()` will always do the array copy, you should extract a
local variable outside the if block.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
+ return path.getDevicePath().matchFullPath(keyPath);
+ } catch (IllegalPathException e) {
+ return false;
+ }
+ };
+ }
+ if (PathPatternUtil.isMultiLevelMatchWildcard(measurement)) {
+ measurementFilter = m -> true;
+ }
+ if (deviceFilter == null) {
+ deviceFilter = d ->
d.toString().equals(path.getDevicePath().getFullPath());
+ }
+
+ if (measurementFilter == null) {
+ measurementFilter = m -> PathPatternUtil.isNodeMatch(measurement,
m.toString());
+ }
+
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (Boolean.TRUE.equals(deviceFilter.apply(device))) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ if (Boolean.TRUE.equals(measurementFilter.apply(entry.getKey()))) {
+ synchronized (entry) {
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+
cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
Review Comment:
make `DataNodeLastCacheManager.invalidateLastCache` return diff size and
then just call
`cacheStats.decreaseMemoryUsage(DataNodeLastCacheManager.invalidateLastCache(cacheEntry))`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
+ return path.getDevicePath().matchFullPath(keyPath);
+ } catch (IllegalPathException e) {
+ return false;
+ }
+ };
+ }
+ if (PathPatternUtil.isMultiLevelMatchWildcard(measurement)) {
+ measurementFilter = m -> true;
+ }
+ if (deviceFilter == null) {
+ deviceFilter = d ->
d.toString().equals(path.getDevicePath().getFullPath());
Review Comment:
directly use `PartialPath.equals`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/TimeSeriesSchemaCache.java:
##########
@@ -393,6 +393,39 @@ public void invalidate(String database) {
dualKeyCache.invalidate(database);
}
+ public void invalidateLastCache(PartialPath path) {
+ if (!path.hasWildcard()) {
+ SchemaCacheEntry entry = dualKeyCache.get(path.getDevicePath(),
path.getMeasurement());
+ if (null == entry) {
+ return;
+ }
+ dualKeyCache.update(
+ new IDualKeyCacheUpdating<PartialPath, String, SchemaCacheEntry>() {
+ @Override
+ public PartialPath getFirstKey() {
+ return path.getDevicePath();
+ }
+
+ @Override
+ public String[] getSecondKeyList() {
+ return new String[] {path.getMeasurement()};
+ }
+
+ @Override
+ public int updateValue(int index, SchemaCacheEntry value) {
+ DataNodeLastCacheManager.invalidateLastCache(value);
+ return 0;
Review Comment:
it should return the diff memory size instead of `0`.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/lastcache/DataNodeLastCacheManager.java:
##########
@@ -65,4 +65,11 @@ public static int updateLastCache(
}
return entry.updateLastCache(timeValuePair, highPriorityUpdate,
latestFlushedTime);
}
+
+ public static void invalidateLastCache(SchemaCacheEntry entry) {
Review Comment:
```suggestion
public static int invalidateLastCache(SchemaCacheEntry entry) {
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
Review Comment:
never do this `new PartialPath(d.toString())`, it's too time consuming!
Actually, for `TimeSeriesSchemaCache`, the first key is always `PartialPath`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
+ return path.getDevicePath().matchFullPath(keyPath);
+ } catch (IllegalPathException e) {
+ return false;
+ }
+ };
+ }
+ if (PathPatternUtil.isMultiLevelMatchWildcard(measurement)) {
+ measurementFilter = m -> true;
+ }
+ if (deviceFilter == null) {
+ deviceFilter = d ->
d.toString().equals(path.getDevicePath().getFullPath());
+ }
+
+ if (measurementFilter == null) {
+ measurementFilter = m -> PathPatternUtil.isNodeMatch(measurement,
m.toString());
+ }
+
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (Boolean.TRUE.equals(deviceFilter.apply(device))) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ if (Boolean.TRUE.equals(measurementFilter.apply(entry.getKey()))) {
+ synchronized (entry) {
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+
cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void invalidateDataRegionLastCache(String database) {
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (device.toString().startsWith(database)) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+ cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/lastcache/ILastCacheContainer.java:
##########
@@ -38,5 +38,8 @@ public interface ILastCacheContainer {
int updateCachedLast(
TimeValuePair timeValuePair, boolean highPriorityUpdate, Long
latestFlushedTime);
+ /** Invalidate Last cache. */
+ void invalidateLastCache();
Review Comment:
```suggestion
int invalidateLastCache();
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java:
##########
@@ -240,6 +243,71 @@ private int evictOneCacheEntry() {
}
}
+ @Override
+ public void invalidateLastCache(PartialPath path) {
+ String measurement = path.getMeasurement();
+ Function<FK, Boolean> deviceFilter = null;
+ Function<SK, Boolean> measurementFilter = null;
+ if (PathPatternUtil.hasWildcard(path.getDevicePath().getFullPath())) {
+ deviceFilter =
+ d -> {
+ try {
+ PartialPath keyPath = new PartialPath(d.toString());
+ return path.getDevicePath().matchFullPath(keyPath);
+ } catch (IllegalPathException e) {
+ return false;
+ }
+ };
+ }
+ if (PathPatternUtil.isMultiLevelMatchWildcard(measurement)) {
+ measurementFilter = m -> true;
+ }
+ if (deviceFilter == null) {
+ deviceFilter = d ->
d.toString().equals(path.getDevicePath().getFullPath());
+ }
+
+ if (measurementFilter == null) {
+ measurementFilter = m -> PathPatternUtil.isNodeMatch(measurement,
m.toString());
+ }
+
+ for (FK device : firstKeyMap.getAllKeys()) {
+ if (Boolean.TRUE.equals(deviceFilter.apply(device))) {
+ ICacheEntryGroup<FK, SK, V, T> entryGroup = firstKeyMap.get(device);
+ for (Iterator<Map.Entry<SK, T>> it = entryGroup.getAllCacheEntries();
it.hasNext(); ) {
+ Map.Entry<SK, T> entry = it.next();
+ if (Boolean.TRUE.equals(measurementFilter.apply(entry.getKey()))) {
+ synchronized (entry) {
+ SchemaCacheEntry cacheEntry = (SchemaCacheEntry)
entry.getValue().getValue();
+ if (cacheEntry.getLastCacheContainer().getCachedLast() != null) {
+ cacheStats.decreaseMemoryUsage(
+
cacheEntry.getLastCacheContainer().getCachedLast().getSize());
+ }
+ DataNodeLastCacheManager.invalidateLastCache(cacheEntry);
Review Comment:
and also what if cacheEntry.getLastCacheContainer() is null, need to
consider this in your new `DataNodeLastCacheManager.invalidateLastCache`
implementation
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java:
##########
@@ -60,6 +60,8 @@ public class DeleteDataNode extends WritePlanNode implements
WALEntryValue {
private final long deleteStartTime;
private final long deleteEndTime;
+ private final boolean isDeleteTimeseries;
Review Comment:
add some comments here to explain why this field doesn't need to be
serialized (because it will always be false except for that in delete
timesereis, and in which case we will set this to true in
`DataNodeInternalRPCServiceImpl.deleteDataForDeleteSchema`.
```suggestion
private final transient boolean isDeleteTimeseries;
```
--
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]
