forked from apache/calcite
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[CALCITE-2040] Create adapter for Apache Arrow
Co-authored-by: Alessandro Solimando <alessandro.solimando@gmail.com> Co-authored-by: Jonathan Swenson <jonathan@exploreomni.com> Co-authored-by: Julian Hyde <jhyde@apache.org> Co-authored-by: Karshit Shah <shahkarshit@yahoo.co.in> Co-authored-by: Michael Mior <mmior@cs.rit.edu>
- Loading branch information
6 people
committed
Mar 11, 2024
1 parent
ad08ce5
commit d4e8830
Showing
34 changed files
with
2,848 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
| dependencies { | ||
| api(project(":core")) | ||
|
|
||
| implementation("com.google.guava:guava") | ||
| implementation("org.apache.arrow:arrow-memory-netty") | ||
| implementation("org.apache.arrow:arrow-vector") | ||
| implementation("org.apache.arrow.gandiva:arrow-gandiva") | ||
| annotationProcessor("org.immutables:value") | ||
| compileOnly("org.immutables:value-annotations") | ||
|
|
||
| testImplementation("org.apache.arrow:arrow-jdbc") | ||
| testImplementation("net.hydromatic:scott-data-hsqldb") | ||
| testImplementation("org.apache.commons:commons-lang3") | ||
| testImplementation(project(":core")) | ||
| testImplementation(project(":testkit")) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| # | ||
| # 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. | ||
| # | ||
| description=Arrow adapter for Calcite | ||
| artifact.name=Calcite Arrow |
82 changes: 82 additions & 0 deletions
82
arrow/src/main/java/org/apache/calcite/adapter/arrow/AbstractArrowEnumerator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * 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.calcite.adapter.arrow; | ||
|
|
||
| import org.apache.calcite.linq4j.Enumerator; | ||
| import org.apache.calcite.util.ImmutableIntList; | ||
| import org.apache.calcite.util.Util; | ||
|
|
||
| import org.apache.arrow.vector.ValueVector; | ||
| import org.apache.arrow.vector.VectorSchemaRoot; | ||
| import org.apache.arrow.vector.VectorUnloader; | ||
| import org.apache.arrow.vector.ipc.ArrowFileReader; | ||
| import org.apache.arrow.vector.ipc.message.ArrowRecordBatch; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Enumerator that reads from a collection of Arrow value-vectors. | ||
| */ | ||
| abstract class AbstractArrowEnumerator implements Enumerator<Object> { | ||
| protected final ArrowFileReader arrowFileReader; | ||
| protected final List<Integer> fields; | ||
| protected final List<ValueVector> valueVectors; | ||
| protected int currRowIndex; | ||
| protected int rowCount; | ||
|
|
||
| AbstractArrowEnumerator(ArrowFileReader arrowFileReader, ImmutableIntList fields) { | ||
| this.arrowFileReader = arrowFileReader; | ||
| this.fields = fields; | ||
| this.valueVectors = new ArrayList<>(fields.size()); | ||
| this.currRowIndex = -1; | ||
| } | ||
|
|
||
| abstract void evaluateOperator(ArrowRecordBatch arrowRecordBatch); | ||
|
|
||
| protected void loadNextArrowBatch() { | ||
| try { | ||
| final VectorSchemaRoot vsr = arrowFileReader.getVectorSchemaRoot(); | ||
| for (int i : fields) { | ||
| this.valueVectors.add(vsr.getVector(i)); | ||
| } | ||
| this.rowCount = vsr.getRowCount(); | ||
| VectorUnloader vectorUnloader = new VectorUnloader(vsr); | ||
| ArrowRecordBatch arrowRecordBatch = vectorUnloader.getRecordBatch(); | ||
| evaluateOperator(arrowRecordBatch); | ||
| } catch (IOException e) { | ||
| throw Util.toUnchecked(e); | ||
| } | ||
| } | ||
|
|
||
| @Override public Object current() { | ||
| if (fields.size() == 1) { | ||
| return this.valueVectors.get(0).getObject(currRowIndex); | ||
| } | ||
| Object[] current = new Object[valueVectors.size()]; | ||
| for (int i = 0; i < valueVectors.size(); i++) { | ||
| ValueVector vector = this.valueVectors.get(i); | ||
| current[i] = vector.getObject(currRowIndex); | ||
| } | ||
| return current; | ||
| } | ||
|
|
||
| @Override public void reset() { | ||
| throw new UnsupportedOperationException(); | ||
| } | ||
| } |
61 changes: 61 additions & 0 deletions
61
arrow/src/main/java/org/apache/calcite/adapter/arrow/ArrowEnumerable.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| /* | ||
| * 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.calcite.adapter.arrow; | ||
|
|
||
| import org.apache.calcite.linq4j.AbstractEnumerable; | ||
| import org.apache.calcite.linq4j.Enumerator; | ||
| import org.apache.calcite.util.ImmutableIntList; | ||
| import org.apache.calcite.util.Util; | ||
|
|
||
| import org.apache.arrow.gandiva.evaluator.Filter; | ||
| import org.apache.arrow.gandiva.evaluator.Projector; | ||
| import org.apache.arrow.vector.ipc.ArrowFileReader; | ||
|
|
||
| import org.checkerframework.checker.nullness.qual.Nullable; | ||
|
|
||
| /** | ||
| * Enumerable that reads from Arrow value-vectors. | ||
| */ | ||
| class ArrowEnumerable extends AbstractEnumerable<Object> { | ||
| private final ArrowFileReader arrowFileReader; | ||
| private final ImmutableIntList fields; | ||
| private final @Nullable Projector projector; | ||
| private final @Nullable Filter filter; | ||
|
|
||
|
|
||
| ArrowEnumerable(ArrowFileReader arrowFileReader, ImmutableIntList fields, | ||
| @Nullable Projector projector, @Nullable Filter filter) { | ||
| this.arrowFileReader = arrowFileReader; | ||
| this.projector = projector; | ||
| this.filter = filter; | ||
| this.fields = fields; | ||
| } | ||
|
|
||
| @Override public Enumerator<Object> enumerator() { | ||
| try { | ||
| if (projector != null) { | ||
| return new ArrowProjectEnumerator(arrowFileReader, fields, projector); | ||
| } else if (filter != null) { | ||
| return new ArrowFilterEnumerator(arrowFileReader, fields, filter); | ||
| } | ||
| throw new IllegalArgumentException( | ||
| "The arrow enumerator must have either a filter or a projection"); | ||
| } catch (Exception e) { | ||
| throw Util.toUnchecked(e); | ||
| } | ||
| } | ||
| } |
102 changes: 102 additions & 0 deletions
102
arrow/src/main/java/org/apache/calcite/adapter/arrow/ArrowFieldType.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| /* | ||
| * 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.calcite.adapter.arrow; | ||
|
|
||
| import org.apache.calcite.adapter.java.JavaTypeFactory; | ||
| import org.apache.calcite.linq4j.tree.Primitive; | ||
| import org.apache.calcite.rel.type.RelDataType; | ||
|
|
||
| import org.apache.arrow.vector.types.FloatingPointPrecision; | ||
| import org.apache.arrow.vector.types.pojo.ArrowType; | ||
|
|
||
| import java.math.BigDecimal; | ||
| import java.util.Date; | ||
| import java.util.List; | ||
|
|
||
| import static java.util.Objects.requireNonNull; | ||
|
|
||
| /** | ||
| * Arrow field type. | ||
| */ | ||
| enum ArrowFieldType { | ||
| INT(Primitive.INT), | ||
| BOOLEAN(Primitive.BOOLEAN), | ||
| STRING(String.class), | ||
| FLOAT(Primitive.FLOAT), | ||
| DOUBLE(Primitive.DOUBLE), | ||
| DATE(Date.class), | ||
| LIST(List.class), | ||
| DECIMAL(BigDecimal.class), | ||
| LONG(Primitive.LONG), | ||
| BYTE(Primitive.BYTE), | ||
| SHORT(Primitive.SHORT); | ||
|
|
||
| private final Class<?> clazz; | ||
|
|
||
| ArrowFieldType(Primitive primitive) { | ||
| this(requireNonNull(primitive.boxClass, "boxClass")); | ||
| } | ||
|
|
||
| ArrowFieldType(Class<?> clazz) { | ||
| this.clazz = clazz; | ||
| } | ||
|
|
||
| public RelDataType toType(JavaTypeFactory typeFactory) { | ||
| RelDataType javaType = typeFactory.createJavaType(clazz); | ||
| RelDataType sqlType = typeFactory.createSqlType(javaType.getSqlTypeName()); | ||
| return typeFactory.createTypeWithNullability(sqlType, true); | ||
| } | ||
|
|
||
| public static ArrowFieldType of(ArrowType arrowType) { | ||
| switch (arrowType.getTypeID()) { | ||
| case Int: | ||
| int bitWidth = ((ArrowType.Int) arrowType).getBitWidth(); | ||
| switch (bitWidth) { | ||
| case 64: | ||
| return LONG; | ||
| case 32: | ||
| return INT; | ||
| case 16: | ||
| return SHORT; | ||
| case 8: | ||
| return BYTE; | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported Int bit width: " + bitWidth); | ||
| } | ||
| case Bool: | ||
| return BOOLEAN; | ||
| case Utf8: | ||
| return STRING; | ||
| case FloatingPoint: | ||
| FloatingPointPrecision precision = ((ArrowType.FloatingPoint) arrowType).getPrecision(); | ||
| switch (precision) { | ||
| case SINGLE: | ||
| return FLOAT; | ||
| case DOUBLE: | ||
| return DOUBLE; | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported Floating point precision: " + precision); | ||
| } | ||
| case Date: | ||
| return DATE; | ||
| case Decimal: | ||
| return DECIMAL; | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported type: " + arrowType); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.