Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.executable.ReferenceCountedJarMetaKeeper;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.commons.udf.UDFInformation;
import org.apache.iotdb.commons.udf.UDFTable;
Expand All @@ -41,7 +42,6 @@
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.udf.api.exception.UDFManagementException;

import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -54,7 +54,6 @@
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
Expand All @@ -67,7 +66,7 @@
ConfigNodeDescriptor.getInstance().getConf();

private final UDFTable udfTable;
private final Map<String, String> existedJarToMD5;
private final ReferenceCountedJarMetaKeeper jarMetaKeeper;

private final UDFExecutableManager udfExecutableManager;

Expand All @@ -77,7 +76,7 @@

public UDFInfo() throws IOException {
udfTable = new UDFTable();
existedJarToMD5 = new HashMap<>();
jarMetaKeeper = new ReferenceCountedJarMetaKeeper();
udfExecutableManager =
UDFExecutableManager.setupAndGetInstance(
CONFIG_NODE_CONF.getUdfTemporaryLibDir(), CONFIG_NODE_CONF.getUdfDir());
Expand All @@ -104,7 +103,8 @@
TSStatusCode.UDF_ALREADY_EXISTS.getStatusCode());
}

if (existedJarToMD5.containsKey(jarName) && !existedJarToMD5.get(jarName).equals(jarMD5)) {
if (jarMetaKeeper.containsJar(jarName)
&& !jarMetaKeeper.jarNameExistsAndMatchesMd5(jarName, jarMD5)) {
throw new IoTDBRuntimeException(
String.format(
ConfigNodeMessages.FAILED_TO_CREATE_UDF_THE_SAME_NAME_JAR_BUT_DIFFERENT,
Expand All @@ -127,15 +127,15 @@
}

public boolean needToSaveJar(String jarName) {
return !existedJarToMD5.containsKey(jarName);
return jarMetaKeeper.needToSaveJar(jarName);
}

public TSStatus addUDFInTable(CreateFunctionPlan physicalPlan) {
try {
final UDFInformation udfInformation = physicalPlan.getUdfInformation();
udfTable.addUDFInformation(udfInformation.getFunctionName(), udfInformation);
if (udfInformation.isUsingURI()) {
existedJarToMD5.put(udfInformation.getJarName(), udfInformation.getJarMD5());
addJarReference(udfInformation.getJarName(), udfInformation.getJarMD5());
if (physicalPlan.getJarFile() != null) {
udfExecutableManager.saveToInstallDir(
ByteBuffer.wrap(physicalPlan.getJarFile().getValues()), udfInformation.getJarName());
Expand Down Expand Up @@ -185,7 +185,10 @@

public TSStatus dropFunction(Model model, String functionName) {
if (udfTable.containsUDF(model, functionName)) {
existedJarToMD5.remove(udfTable.getUDFInformation(model, functionName).getJarName());
final UDFInformation udfInformation = udfTable.getUDFInformation(model, functionName);
if (udfInformation.isUsingURI()) {
removeJarReference(udfInformation.getJarName());
}
udfTable.removeUDFInformation(model, functionName);
}
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
Expand All @@ -204,7 +207,7 @@

@TestOnly
public Map<String, String> getRawExistedJarToMD5() {
return existedJarToMD5;
return jarMetaKeeper.getJarNameToMd5Map();
}

@Override
Expand Down Expand Up @@ -248,30 +251,39 @@
deserializeExistedJarToMD5(fileInputStream);

udfTable.deserializeUDFTable(fileInputStream);
rebuildJarMetadataFromUDFTable();
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After deserializeExistedJarToMD5, rebuildJarMetadataFromUDFTable() clears both maps and rebuilds from udfTable. That makes udfTable the source of truth on load.

A short comment here explaining that deserialized existedJarToMD5 is intentionally discarded would help future readers.

} finally {
releaseUDFTableLock();
}
}

public void serializeExistedJarToMD5(OutputStream outputStream) throws IOException {
ReadWriteIOUtils.write(existedJarToMD5.size(), outputStream);
for (Map.Entry<String, String> entry : existedJarToMD5.entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
}
jarMetaKeeper.serializeJarNameToMd5(outputStream);
}

public void deserializeExistedJarToMD5(InputStream inputStream) throws IOException {
int size = ReadWriteIOUtils.readInt(inputStream);
while (size > 0) {
existedJarToMD5.put(
ReadWriteIOUtils.readString(inputStream), ReadWriteIOUtils.readString(inputStream));
size--;
}
jarMetaKeeper.deserializeJarNameToMd5(inputStream);
}

public void clear() {
existedJarToMD5.clear();
jarMetaKeeper.clear();
udfTable.clear();
}

private void addJarReference(String jarName, String jarMD5) {
jarMetaKeeper.addReference(jarName, jarMD5);
}

private void removeJarReference(String jarName) {
jarMetaKeeper.removeReference(jarName);
}

private void rebuildJarMetadataFromUDFTable() {

Check warning on line 281 in iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Abbreviation in name 'rebuildJarMetadataFromUDFTable' must contain no more than '2' consecutive capital letters.

See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5Imge8RiG7CBVeO-Sc&open=AZ5Imge8RiG7CBVeO-Sc&pullRequest=17732
jarMetaKeeper.clear();
for (UDFInformation udfInformation : udfTable.getAllInformationList()) {
if (udfInformation.isUsingURI()) {
addJarReference(udfInformation.getJarName(), udfInformation.getJarMD5());
}
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.iotdb.common.rpc.thrift.FunctionType;
import org.apache.iotdb.common.rpc.thrift.Model;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.udf.UDFInformation;
import org.apache.iotdb.commons.udf.UDFType;
import org.apache.iotdb.confignode.consensus.request.write.function.CreateFunctionPlan;
Expand All @@ -41,6 +42,10 @@

public class UDFInfoTest {

private static final String SHARED_JAR_NAME = "shared.jar";
private static final String SHARED_JAR_MD5 = "12345";
private static final String DIFFERENT_JAR_MD5 = "54321";

private static UDFInfo udfInfo;
private static UDFInfo udfInfoSaveBefore;
private static final File snapshotDir = new File(BASE_OUTPUT_PATH, "snapshot");
Expand All @@ -63,37 +68,74 @@ public static void cleanup() throws IOException {
}

@Test
public void testSnapshot() throws TException, IOException, IllegalPathException {
UDFInformation udfInformation =
new UDFInformation(
"test1",
"test1",
UDFType.of(Model.TREE, FunctionType.NONE, true),
true,
"test1.jar",
"12345");
CreateFunctionPlan createFunctionPlan =
new CreateFunctionPlan(udfInformation, new Binary(new byte[] {1, 2, 3}));
udfInfo.addUDFInTable(createFunctionPlan);
udfInfoSaveBefore.addUDFInTable(createFunctionPlan);

udfInformation =
new UDFInformation(
"test2",
"test2",
UDFType.of(Model.TREE, FunctionType.NONE, true),
true,
"test2.jar",
"123456");
createFunctionPlan = new CreateFunctionPlan(udfInformation, new Binary(new byte[] {1, 2, 3}));
udfInfo.addUDFInTable(createFunctionPlan);
udfInfoSaveBefore.addUDFInTable(createFunctionPlan);
public void testDropOneSharedJarReferenceKeepsJarMetadata()
throws TException, IOException, IllegalPathException {
clearUdfInfos();

udfInfo.addUDFInTable(createFunctionPlan("test1", SHARED_JAR_NAME, SHARED_JAR_MD5, true));
udfInfo.addUDFInTable(createFunctionPlan("test2", SHARED_JAR_NAME, SHARED_JAR_MD5, false));

udfInfo.dropFunction(Model.TREE, "test1");

Assert.assertFalse(udfInfo.needToSaveJar(SHARED_JAR_NAME));
Assert.assertEquals(1, udfInfo.getRawExistedJarToMD5().size());
Assert.assertEquals(SHARED_JAR_MD5, udfInfo.getRawExistedJarToMD5().get(SHARED_JAR_NAME));

udfInfo.validate(Model.TREE, "test3", SHARED_JAR_NAME, SHARED_JAR_MD5);
try {
udfInfo.validate(Model.TREE, "test3", SHARED_JAR_NAME, DIFFERENT_JAR_MD5);
Assert.fail("Expected shared jar conflict after dropping only one referenced UDF.");
} catch (IoTDBRuntimeException e) {
Assert.assertEquals(
org.apache.iotdb.rpc.TSStatusCode.UDF_ALREADY_EXISTS.getStatusCode(), e.getErrorCode());
}
}

@Test
public void testSnapshotRebuildsSharedJarReferences()
throws TException, IOException, IllegalPathException {
clearUdfInfos();
FileUtils.cleanDirectory(snapshotDir);

CreateFunctionPlan createFunctionPlan1 =
createFunctionPlan("test1", SHARED_JAR_NAME, SHARED_JAR_MD5, true);
CreateFunctionPlan createFunctionPlan2 =
createFunctionPlan("test2", SHARED_JAR_NAME, SHARED_JAR_MD5, false);

udfInfo.addUDFInTable(createFunctionPlan1);
udfInfo.addUDFInTable(createFunctionPlan2);
udfInfoSaveBefore.addUDFInTable(createFunctionPlan1);
udfInfoSaveBefore.addUDFInTable(createFunctionPlan2);

udfInfo.processTakeSnapshot(snapshotDir);
udfInfo.clear();
udfInfo.processLoadSnapshot(snapshotDir);

Assert.assertEquals(udfInfoSaveBefore.getRawExistedJarToMD5(), udfInfo.getRawExistedJarToMD5());
Assert.assertEquals(udfInfoSaveBefore.getRawUDFTable(), udfInfo.getRawUDFTable());

udfInfo.dropFunction(Model.TREE, "test1");
Assert.assertFalse(udfInfo.needToSaveJar(SHARED_JAR_NAME));
Assert.assertEquals(SHARED_JAR_MD5, udfInfo.getRawExistedJarToMD5().get(SHARED_JAR_NAME));
}

private static void clearUdfInfos() {
udfInfo.clear();
udfInfoSaveBefore.clear();
}

private static CreateFunctionPlan createFunctionPlan(
String functionName, String jarName, String jarMD5, boolean includeJarFile)
throws IllegalPathException {
UDFInformation udfInformation =
new UDFInformation(
functionName,
functionName,
UDFType.of(Model.TREE, FunctionType.NONE, true),
true,
jarName,
jarMD5);
return new CreateFunctionPlan(
udfInformation, includeJarFile ? new Binary(new byte[] {1, 2, 3}) : null);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
/*
* 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.iotdb.commons.executable;

import org.apache.tsfile.utils.ReadWriteIOUtils;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

public class ReferenceCountedJarMetaKeeper {

private final Map<String, String> jarNameToMd5Map = new HashMap<>();
private final Map<String, Integer> jarNameToReferenceCountMap = new HashMap<>();

public synchronized boolean containsJar(final String jarName) {
return jarNameToMd5Map.containsKey(jarName);
}

public synchronized boolean needToSaveJar(final String jarName) {
return !containsJar(jarName);
}

public synchronized boolean jarNameExistsAndMatchesMd5(final String jarName, final String md5) {
return containsJar(jarName) && Objects.equals(jarNameToMd5Map.get(jarName), md5);
}

public synchronized void addReference(final String jarName, final String md5) {
if (jarNameToReferenceCountMap.containsKey(jarName)) {

Check warning on line 49 in iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/executable/ReferenceCountedJarMetaKeeper.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace this "Map.containsKey()" with a call to "Map.computeIfPresent()".

See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ5yqIlh8I4eVE1_iY3I&open=AZ5yqIlh8I4eVE1_iY3I&pullRequest=17732
jarNameToReferenceCountMap.put(jarName, jarNameToReferenceCountMap.get(jarName) + 1);
return;
}

jarNameToReferenceCountMap.put(jarName, 1);
jarNameToMd5Map.put(jarName, md5);
}

public synchronized void removeReference(final String jarName) {
final Integer referenceCount = jarNameToReferenceCountMap.get(jarName);
if (referenceCount == null || referenceCount <= 1) {
jarNameToReferenceCountMap.remove(jarName);
jarNameToMd5Map.remove(jarName);
return;
}

jarNameToReferenceCountMap.put(jarName, referenceCount - 1);
}

public synchronized void clear() {
jarNameToMd5Map.clear();
jarNameToReferenceCountMap.clear();
}

public synchronized Map<String, String> getJarNameToMd5Map() {
return new HashMap<>(jarNameToMd5Map);
}

public synchronized void serializeJarNameToMd5(final OutputStream outputStream)
throws IOException {
ReadWriteIOUtils.write(jarNameToMd5Map.size(), outputStream);
for (final Map.Entry<String, String> entry : jarNameToMd5Map.entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
}
}

public synchronized void deserializeJarNameToMd5(final InputStream inputStream)
throws IOException {
clear();

int size = ReadWriteIOUtils.readInt(inputStream);
while (size > 0) {
addReference(
ReadWriteIOUtils.readString(inputStream), ReadWriteIOUtils.readString(inputStream));
size--;
}
}

public synchronized void serializeJarNameToMd5AndReferenceCount(final OutputStream outputStream)
throws IOException {
ReadWriteIOUtils.write(jarNameToMd5Map.size(), outputStream);
for (final Map.Entry<String, String> entry : jarNameToMd5Map.entrySet()) {
final String jarName = entry.getKey();
ReadWriteIOUtils.write(jarName, outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
ReadWriteIOUtils.write(jarNameToReferenceCountMap.getOrDefault(jarName, 0), outputStream);
}
}

public synchronized void deserializeJarNameToMd5AndReferenceCount(final InputStream inputStream)
throws IOException {
clear();

final int jarSize = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < jarSize; i++) {
final String jarName = ReadWriteIOUtils.readString(inputStream);
jarNameToMd5Map.put(jarName, ReadWriteIOUtils.readString(inputStream));
jarNameToReferenceCountMap.put(jarName, ReadWriteIOUtils.readInt(inputStream));
}
}
}
Loading
Loading