Extracted vector-type from sqrl project

Signed-off-by: Marvin Froeder <marvin@datasqrl.com>

Author: velo

types/pom.xml

@@ -31,6 +31,7 @@
 
   <modules>
     <module>json-type</module>
+    <module>vector-type</module>
   </modules>
 
 </project>

types/vector-type/pom.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright © 2024 DataSQRL (contact@datasqrl.com)
+
+    Licensed 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.datasqrl.flinkrunner</groupId>
+    <artifactId>types</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>vector-type</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-common</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+  </dependencies>
+</project>

types/vector-type/src/main/java/com/datasqrl/types/vector/FlinkVectorType.java

@@ -0,0 +1,19 @@
+package com.datasqrl.types.vector;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+
+@DataTypeHint(
+    value = "RAW",
+    bridgedTo = FlinkVectorType.class,
+    rawSerializer = FlinkVectorTypeSerializer.class)
+public class FlinkVectorType {
+  public double[] value;
+
+  public FlinkVectorType(double[] value) {
+    this.value = value;
+  }
+
+  public double[] getValue() {
+    return value;
+  }
+}

types/vector-type/src/main/java/com/datasqrl/types/vector/FlinkVectorTypeSerializer.java

@@ -0,0 +1,84 @@
+package com.datasqrl.types.vector;
+
+import java.io.IOException;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+public class FlinkVectorTypeSerializer extends TypeSerializer<FlinkVectorType> {
+
+  @Override
+  public boolean isImmutableType() {
+    return true;
+  }
+
+  @Override
+  public FlinkVectorType createInstance() {
+    return new FlinkVectorType(null);
+  }
+
+  @Override
+  public FlinkVectorType copy(FlinkVectorType from) {
+    return new FlinkVectorType(from.getValue());
+  }
+
+  @Override
+  public FlinkVectorType copy(FlinkVectorType from, FlinkVectorType reuse) {
+    return copy(from);
+  }
+
+  @Override
+  public int getLength() {
+    return -1; // indicates that this serializer does not have a fixed length
+  }
+
+  @Override
+  public void serialize(FlinkVectorType record, DataOutputView target) throws IOException {
+    target.writeInt(record.getValue().length); // First write the length of the array
+    for (double v : record.getValue()) {
+      target.writeDouble(v); // Write each double value
+    }
+  }
+
+  @Override
+  public FlinkVectorType deserialize(DataInputView source) throws IOException {
+    int length = source.readInt();
+    double[] array = new double[length];
+    for (int i = 0; i < length; i++) {
+      array[i] = source.readDouble();
+    }
+    return new FlinkVectorType(array);
+  }
+
+  @Override
+  public FlinkVectorType deserialize(FlinkVectorType reuse, DataInputView source)
+      throws IOException {
+    return deserialize(source);
+  }
+
+  @Override
+  public void copy(DataInputView source, DataOutputView target) throws IOException {
+    target.writeUTF(source.readUTF());
+  }
+
+  @Override
+  public TypeSerializer<FlinkVectorType> duplicate() {
+    return this;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    return obj instanceof FlinkVectorTypeSerializer;
+  }
+
+  @Override
+  public int hashCode() {
+    return FlinkVectorTypeSerializer.class.hashCode();
+  }
+
+  @Override
+  public TypeSerializerSnapshot<FlinkVectorType> snapshotConfiguration() {
+    return new FlinkVectorTypeSerializerSnapshot();
+  }
+}

types/vector-type/src/main/java/com/datasqrl/types/vector/FlinkVectorTypeSerializerSnapshot.java

@@ -0,0 +1,59 @@
+package com.datasqrl.types.vector;
+
+import java.io.IOException;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+public class FlinkVectorTypeSerializerSnapshot implements TypeSerializerSnapshot<FlinkVectorType> {
+
+  private Class<FlinkVectorTypeSerializer> serializerClass;
+
+  public FlinkVectorTypeSerializerSnapshot() {
+    this.serializerClass = FlinkVectorTypeSerializer.class;
+  }
+
+  @Override
+  public int getCurrentVersion() {
+    return 1;
+  }
+
+  @Override
+  public void writeSnapshot(DataOutputView out) throws IOException {
+    out.writeUTF(FlinkVectorTypeSerializer.class.getName());
+  }
+
+  @Override
+  public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+      throws IOException {
+    String className = in.readUTF();
+    try {
+      this.serializerClass =
+          (Class<FlinkVectorTypeSerializer>) Class.forName(className, true, userCodeClassLoader);
+    } catch (ClassNotFoundException e) {
+      throw new IOException("Failed to find serializer class: " + className, e);
+    }
+  }
+
+  @Override
+  public TypeSerializer restoreSerializer() {
+    try {
+      return serializerClass.newInstance();
+    } catch (InstantiationException | IllegalAccessException e) {
+      throw new RuntimeException(
+          "Failed to instantiate serializer class: " + serializerClass.getName(), e);
+    }
+  }
+
+  @Override
+  public TypeSerializerSchemaCompatibility resolveSchemaCompatibility(
+      TypeSerializer newSerializer) {
+    if (newSerializer.getClass() == this.serializerClass) {
+      return TypeSerializerSchemaCompatibility.compatibleAsIs();
+    } else {
+      return TypeSerializerSchemaCompatibility.incompatible();
+    }
+  }
+}