This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch object_type
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/object_type by this push:
new 48520edda26 add GeoPenetrate function
new 1bc950dc155 Merge branch 'object_type' of
https://github.com/apache/iotdb into object_type
48520edda26 is described below
commit 48520edda261427fcd6c663052c7ce681216ad38
Author: JackieTien97 <[email protected]>
AuthorDate: Mon Jul 7 19:50:31 2025 +0800
add GeoPenetrate function
---
.../main/java/org/apache/iotdb/ObjectExample.java | 8 +-
.../java/org/apache/iotdb/ObjectReadExample.java | 16 +++
.../java/org/apache/iotdb/udf/api/type/Type.java | 10 +-
.../relational/function/scalar/GeoPenetrate.java | 123 +++++++++++++++++++++
.../unary/scalar/ReadObjectColumnTransformer.java | 2 +-
.../utils/model/CompressedTsFileModelReader.java | 2 +-
.../apache/iotdb/db/utils/model/ModelReader.java | 22 ++--
.../iotdb/db/utils/model/ModelReaderType.java | 25 -----
.../utils/model/UnCompressedTiffModelReader.java | 2 +-
.../commons/udf/utils/UDFDataTypeTransformer.java | 7 ++
10 files changed, 177 insertions(+), 40 deletions(-)
diff --git a/example/session/src/main/java/org/apache/iotdb/ObjectExample.java
b/example/session/src/main/java/org/apache/iotdb/ObjectExample.java
index b040dc5313b..6a880cc0174 100644
--- a/example/session/src/main/java/org/apache/iotdb/ObjectExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/ObjectExample.java
@@ -48,7 +48,7 @@ public class ObjectExample {
.username("root")
.password("root")
.build()) {
- session.executeNonQueryStatement("CREATE DATABASE test1");
+ session.executeNonQueryStatement("CREATE DATABASE IF NOT EXISTS test1");
session.executeNonQueryStatement("use test1");
// insert table data by tablet
@@ -82,7 +82,8 @@ public class ObjectExample {
true,
0,
Files.readAllBytes(
-
Paths.get("/Users/ht/Downloads/2_1746622362350_fa24aa15233f4e76bcda789a5771f43f")));
+ Paths.get(
+
"/Users/jackietien/Downloads/2_1746622362350_fa24aa15233f4e76bcda789a5771f43f")));
session.insert(tablet);
tablet.reset();
@@ -99,7 +100,8 @@ public class ObjectExample {
true,
0,
Files.readAllBytes(
-
Paths.get("/Users/ht/Downloads/2_1746622367063_8fb5ac8e21724140874195b60b878664")));
+ Paths.get(
+
"/Users/jackietien/Downloads/2_1746622367063_8fb5ac8e21724140874195b60b878664")));
session.insert(tablet);
tablet.reset();
diff --git
a/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java
b/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java
index 4fd68322fe0..bf4417c15be 100644
--- a/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.session.TableSessionBuilder;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.tsfile.utils.Binary;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.Collections;
public class ObjectReadExample {
@@ -71,6 +73,20 @@ public class ObjectReadExample {
}
}
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("select geo_penetrate(file,
'0,3,7501,7504') from test1")) {
+ SessionDataSet.DataIterator iterator = dataSet.iterator();
+ while (iterator.next()) {
+ Binary binary = iterator.getBlob(1);
+ ByteBuffer byteBuffer = ByteBuffer.wrap(binary.getValues());
+ float[] res = new float[byteBuffer.limit() / Float.BYTES];
+ for (int i = 0; i < res.length; i++) {
+ res[i] = byteBuffer.getFloat();
+ }
+ System.out.println(Arrays.toString(res));
+ }
+ }
+
} catch (IoTDBConnectionException e) {
e.printStackTrace();
} catch (StatementExecutionException e) {
diff --git
a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/Type.java
b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/Type.java
index a5c2852ac1b..c4c38c285dc 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/Type.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/type/Type.java
@@ -56,7 +56,11 @@ public enum Type {
BLOB((byte) 10),
/* STRING */
- STRING((byte) 11);
+ STRING((byte) 11),
+
+ /* OBJECT */
+ OBJECT((byte) 12);
+
private final byte dataType;
Type(byte type) {
@@ -92,6 +96,7 @@ public enum Type {
case DATE:
return o instanceof LocalDate;
case BLOB:
+ case OBJECT:
return o instanceof Binary;
case STRING:
case TEXT:
@@ -102,7 +107,8 @@ public enum Type {
}
public static List<Type> allTypes() {
- return Arrays.asList(BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT,
TIMESTAMP, DATE, BLOB, STRING);
+ return Arrays.asList(
+ BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT, TIMESTAMP, DATE, BLOB,
STRING, OBJECT);
}
public static List<Type> numericTypes() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/scalar/GeoPenetrate.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/scalar/GeoPenetrate.java
new file mode 100644
index 00000000000..440b3728bc4
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/scalar/GeoPenetrate.java
@@ -0,0 +1,123 @@
+/*
+ * 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.db.queryengine.plan.relational.function.scalar;
+
+import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.utils.model.ModelReader;
+import org.apache.iotdb.udf.api.customizer.analysis.ScalarFunctionAnalysis;
+import org.apache.iotdb.udf.api.customizer.parameter.FunctionArguments;
+import org.apache.iotdb.udf.api.exception.UDFArgumentNotValidException;
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import static
org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.ReadObjectColumnTransformer.getObjectPathFromBinary;
+
+public class GeoPenetrate implements ScalarFunction {
+
+ private ModelReader reader;
+ private List<List<Integer>> startAndEndTimeArray;
+
+ @Override
+ public ScalarFunctionAnalysis analyze(FunctionArguments arguments)
+ throws UDFArgumentNotValidException {
+ if (arguments.getArgumentsSize() != 2 && arguments.getArgumentsSize() !=
3) {
+ throw new UDFArgumentNotValidException(
+ "function requires 2 or 3 arguments, current parameters' size is "
+ + arguments.getArgumentsSize());
+ }
+ if (arguments.getDataType(0) != Type.OBJECT) {
+ throw new UDFArgumentNotValidException("function's first parameter type
should be OBJECT");
+ }
+
+ if (arguments.getDataType(1) != Type.STRING) {
+ throw new UDFArgumentNotValidException("function's second parameter type
should be STRING");
+ }
+
+ if (arguments.getArgumentsSize() == 3 && arguments.getDataType(1) !=
Type.STRING) {
+ throw new UDFArgumentNotValidException("function's third parameter type
should be STRING");
+ }
+
+ return new
ScalarFunctionAnalysis.Builder().outputDataType(Type.BLOB).build();
+ }
+
+ @Override
+ public void beforeStart(FunctionArguments arguments) throws UDFException {
+ if (arguments.getArgumentsSize() == 2) {
+ reader = ModelReader.getDefaultInstance();
+ }
+ }
+
+ @Override
+ public Object evaluate(Record input) throws UDFException {
+ if (reader == null) {
+ reader = ModelReader.getInstance(input.getString(2));
+ }
+ if (startAndEndTimeArray == null) {
+ startAndEndTimeArray = getStartAndEndTimeArray(input.getString(1));
+ }
+ if (input.isNull(0)) {
+ return null;
+ } else {
+ File objectPath = getObjectPathFromBinary(input.getBinary(0));
+ List<float[]> res = reader.penetrate(objectPath.getAbsolutePath(),
startAndEndTimeArray);
+ int count = 0;
+ for (float[] array : res) {
+ count += array.length;
+ }
+ ByteBuffer byteBuffer = ByteBuffer.allocate(count * Float.BYTES);
+ for (float[] array : res) {
+ for (float value : array) {
+ ReadWriteIOUtils.write(value, byteBuffer);
+ }
+ }
+ return new Binary(byteBuffer.array());
+ }
+ }
+
+ // argument like: 1,2;200,201;300,401
+ private List<List<Integer>> getStartAndEndTimeArray(String argument) {
+ String[] arguments = argument.split(",");
+ if (arguments.length % 2 != 0) {
+ throw new SemanticException("second argument is wrong: " + argument);
+ }
+ List<List<Integer>> res = new ArrayList<>(arguments.length);
+ for (int index = 0; index < arguments.length; index += 2) {
+ List<Integer> startAndEndTime = new ArrayList<>(2);
+ try {
+ startAndEndTime.add(Integer.parseInt(arguments[index]));
+ startAndEndTime.add(Integer.parseInt(arguments[index + 1]));
+ } catch (NumberFormatException e) {
+ throw new SemanticException("second argument is wrong: " + argument);
+ }
+ res.add(startAndEndTime);
+ }
+ return res;
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java
index 74db2d31ca3..60a178a33c8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java
@@ -130,7 +130,7 @@ public class ReadObjectColumnTransformer extends
UnaryColumnTransformer {
return new Binary(bytes);
}
- private File getObjectPathFromBinary(Binary binary) {
+ public static File getObjectPathFromBinary(Binary binary) {
byte[] bytes = binary.getValues();
String relativeObjectFilePath =
new String(bytes, 8, bytes.length - 8, TSFileConfig.STRING_CHARSET);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/CompressedTsFileModelReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/CompressedTsFileModelReader.java
index 8cc0c4a9eb6..2849e0657dd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/CompressedTsFileModelReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/CompressedTsFileModelReader.java
@@ -48,7 +48,7 @@ public class CompressedTsFileModelReader extends ModelReader {
private static final Logger LOGGER =
LoggerFactory.getLogger(CompressedTsFileModelReader.class);
@Override
- List<float[]> penetrate(String filePath, List<List<Integer>>
startAndEndTimeArray) {
+ public List<float[]> penetrate(String filePath, List<List<Integer>>
startAndEndTimeArray) {
try {
List<float[]> results = new ArrayList<>(startAndEndTimeArray.size());
int currentQueryIndex = 0;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReader.java
index 14b4f939323..18a6f5aed43 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReader.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.utils.model;
+import org.apache.iotdb.db.exception.sql.SemanticException;
+
import java.util.List;
public abstract class ModelReader {
@@ -30,14 +32,20 @@ public abstract class ModelReader {
* integers
* @return a list of float arrays, each array corresponds to a start and end
time pair
*/
- abstract List<float[]> penetrate(String filePath, List<List<Integer>>
startAndEndTimeArray);
+ public abstract List<float[]> penetrate(
+ String filePath, List<List<Integer>> startAndEndTimeArray);
+
+ public static ModelReader getDefaultInstance() {
+ return new CompressedTsFileModelReader();
+ }
- public ModelReader getInstance(ModelReaderType modelFileType) {
- switch (modelFileType) {
- case UNCOMPRESSED_TIFF:
- return new UnCompressedTiffModelReader();
- default:
- return new CompressedTsFileModelReader();
+ public static ModelReader getInstance(String modelFileType) {
+ if (modelFileType.equalsIgnoreCase("UNCOMPRESSED_TIFF")) {
+ return new UnCompressedTiffModelReader();
+ } else if (modelFileType.equalsIgnoreCase("COMPRESSED_TSFILE")) {
+ return new CompressedTsFileModelReader();
+ } else {
+ throw new SemanticException("Unsupported model file type: " +
modelFileType);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReaderType.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReaderType.java
deleted file mode 100644
index fe07b4d0f8e..00000000000
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/ModelReaderType.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.db.utils.model;
-
-public enum ModelReaderType {
- COMPRESSED_TSFILE,
- UNCOMPRESSED_TIFF
-}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/UnCompressedTiffModelReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/UnCompressedTiffModelReader.java
index 99f3d183126..debfeab42a3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/UnCompressedTiffModelReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/model/UnCompressedTiffModelReader.java
@@ -25,7 +25,7 @@ import java.util.List;
public class UnCompressedTiffModelReader extends ModelReader {
@Override
- List<float[]> penetrate(String filePath, List<List<Integer>>
startAndEndTimeArray) {
+ public List<float[]> penetrate(String filePath, List<List<Integer>>
startAndEndTimeArray) {
return Collections.emptyList();
}
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
index 83792ebf66a..cf9ecf89f96 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
@@ -30,6 +30,7 @@ import org.apache.tsfile.read.common.type.DoubleType;
import org.apache.tsfile.read.common.type.FloatType;
import org.apache.tsfile.read.common.type.IntType;
import org.apache.tsfile.read.common.type.LongType;
+import org.apache.tsfile.read.common.type.ObjectType;
import org.apache.tsfile.read.common.type.StringType;
import org.apache.tsfile.read.common.type.TimestampType;
@@ -82,6 +83,8 @@ public class UDFDataTypeTransformer {
return Type.BLOB;
case STRING:
return Type.STRING;
+ case OBJECT:
+ return Type.OBJECT;
default:
throw new IllegalArgumentException("Invalid input: " + type);
}
@@ -112,6 +115,8 @@ public class UDFDataTypeTransformer {
return BlobType.BLOB;
case STRING:
return StringType.STRING;
+ case OBJECT:
+ return ObjectType.OBJECT;
default:
throw new IllegalArgumentException("Invalid input: " + type);
}
@@ -139,6 +144,8 @@ public class UDFDataTypeTransformer {
return Type.BLOB;
case 11:
return Type.STRING;
+ case 12:
+ return Type.OBJECT;
default:
throw new IllegalArgumentException("Invalid input: " + type);
}
