+ * 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.flink.table.runtime.arrow.vectors; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.dataformat.vector.LongColumnVector; +import org.apache.flink.util.Preconditions; + +import org.apache.arrow.vector.BigIntVector; + +/** + * Arrow column vector for BigInt. + */ +@Internal +public final class ArrowBigIntColumnVector implements LongColumnVector { + + /** + * Container which is used to store the sequence of bigint values of a column to read. + */ + private final BigIntVector bigIntVector; + + public ArrowBigIntColumnVector(BigIntVector bigIntVector) { + this.bigIntVector = Preconditions.checkNotNull(bigIntVector); + } + + @Override + public long getLong(int i) { + return bigIntVector.get(i); + } + + @Override + public boolean isNullAt(int i) { + return bigIntVector.isNull(i); + } + + @Override + public void reset() { + bigIntVector.reset(); + } +} diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowIntColumnVector.java b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowIntColumnVector.java new file mode 100644 index 0000000000000..5673d9f1abf0b --- /dev/null +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowIntColumnVector.java @@ -0,0 +1,53 @@ +/* + * 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.flink.table.runtime.arrow.vectors; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.dataformat.vector.IntColumnVector; +import org.apache.flink.util.Preconditions; + +import org.apache.arrow.vector.IntVector; + +/** + * Arrow column vector for Int. + */ +@Internal +public final class ArrowIntColumnVector implements IntColumnVector { + + private final IntVector intVector; + + public ArrowIntColumnVector(IntVector intVector) { + this.intVector = Preconditions.checkNotNull(intVector); + } + + @Override + public int getInt(int i) { + return intVector.get(i); + } + + @Override + public boolean isNullAt(int i) { + return intVector.isNull(i); + } + + @Override + public void reset() { + intVector.reset(); + } +} diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowSmallIntColumnVector.java b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowSmallIntColumnVector.java new file mode 100644 index 0000000000000..83bcdc119a5dd --- /dev/null +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowSmallIntColumnVector.java @@ -0,0 +1,53 @@ +/* + * 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.flink.table.runtime.arrow.vectors; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.dataformat.vector.ShortColumnVector; +import org.apache.flink.util.Preconditions; + +import org.apache.arrow.vector.SmallIntVector; + +/** + * Arrow column vector for Int. + */ +@Internal +public final class ArrowSmallIntColumnVector implements ShortColumnVector { + + private final SmallIntVector smallIntVector; + + public ArrowSmallIntColumnVector(SmallIntVector smallIntVector) { + this.smallIntVector = Preconditions.checkNotNull(smallIntVector); + } + + @Override + public short getShort(int i) { + return smallIntVector.get(i); + } + + @Override + public boolean isNullAt(int i) { + return smallIntVector.isNull(i); + } + + @Override + public void reset() { + smallIntVector.reset(); + } +} diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowTinyIntColumnVector.java b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowTinyIntColumnVector.java new file mode 100644 index 0000000000000..11225968ca0d3 --- /dev/null +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/ArrowTinyIntColumnVector.java @@ -0,0 +1,53 @@ +/* + * 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.flink.table.runtime.arrow.vectors;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.dataformat.vector.ByteColumnVector;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.arrow.vector.TinyIntVector;
+
+/**
+ * Arrow column vector for TinyInt.
+ */
+@Internal
+public final class ArrowTinyIntColumnVector implements ByteColumnVector {
+
+ private final TinyIntVector tinyIntVector;
+
+ public ArrowTinyIntColumnVector(TinyIntVector tinyIntVector) {
+ this.tinyIntVector = Preconditions.checkNotNull(tinyIntVector);
+ }
+
+ @Override
+ public byte getByte(int i) {
+ return tinyIntVector.get(i);
+ }
+
+ @Override
+ public boolean isNullAt(int i) {
+ return tinyIntVector.isNull(i);
+ }
+
+ @Override
+ public void reset() {
+ tinyIntVector.reset();
+ }
+}
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/BaseRowArrowReader.java b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/BaseRowArrowReader.java
new file mode 100644
index 0000000000000..67dd06b9486a8
--- /dev/null
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/arrow/vectors/BaseRowArrowReader.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.table.runtime.arrow.vectors;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.ColumnarRow;
+import org.apache.flink.table.dataformat.vector.ColumnVector;
+import org.apache.flink.table.dataformat.vector.VectorizedColumnBatch;
+import org.apache.flink.table.runtime.arrow.ArrowReader;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * {@link ArrowReader} which read the underlying Arrow format data as {@link BaseRow}.
+ */
+@Internal
+public final class BaseRowArrowReader implements ArrowReader