Thoughts

Author: Martin Traverse

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/ArrowToAvroUtils.java

@@ -0,0 +1,142 @@
+/*
+ * 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.arrow.adapter.avro;
+
+import org.apache.arrow.adapter.avro.producers.*;
+import org.apache.arrow.adapter.avro.producers.logical.*;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.*;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+public class ArrowToAvroUtils {
+
+  static Schema convertSchema(org.apache.arrow.vector.types.pojo.Schema arrowSchema) {
+
+    SchemaBuilder.FieldAssembler<?> fieldAssembler = SchemaBuilder.record("").fields();
+
+    for (Field arrowField : arrowSchema.getFields()) {
+      fieldAssembler = convertFieldSchema(fieldAssembler, arrowField);
+    }
+
+    return fieldAssembler.requiredBytes().endRecord();
+  }
+
+  static SchemaBuilder.FieldAssembler<?> convertFieldSchema(
+          SchemaBuilder.FieldAssembler<?> builder,
+          Field arrowField) {
+
+    return builder.
+  }
+
+  static CompositeAvroProducer createCompositeProducer(Schema schema) {
+
+    Preconditions.checkNotNull(schema, "Avro schema object can't be null");
+
+    List<Producer<?>> producers = new ArrayList<>();
+    final Set<String> skipFieldNames = Set.of();  // TODO: Is this needed?
+
+    Schema.Type type = schema.getType();
+    if (type == Schema.Type.RECORD) {
+      for (Schema.Field field : schema.getFields()) {
+        if (skipFieldNames.contains(field.name())) {
+//          producers.add(createSkipConsumer(field.schema()));  // TODO
+        } else {
+          Producer<?> producer = createProducer(field.schema(), field.name(), config);
+          producers.add(producer);
+        }
+      }
+    } else {
+      Producer<?> producer = createProducer(schema, "", config);
+      producers.add(producer);
+    }
+
+    return new CompositeAvroProducer(producers);
+  }
+
+
+  private static BaseAvroProducer<?> createProducer(Schema schema, FieldVector vector) {
+
+    Preconditions.checkNotNull(schema, "Avro schema object can't be null");
+
+    final Schema.Type type = schema.getType();
+    final LogicalType logicalType = schema.getLogicalType();
+
+    switch (type) {
+      case UNION:
+        return createUnionProducer(schema, vector);
+      case ARRAY:
+        return createArrayProducer(schema, vector);
+      case MAP:
+        return createMapProducer(schema, vector);
+      case RECORD:
+        return createStructProducer(schema, vector);
+      case ENUM:
+        return new AvroEnumProducer((VarCharVector) vector);
+      case STRING:
+        return new AvroStringProducer((VarCharVector) vector);
+      case FIXED:
+        if (logicalType instanceof LogicalTypes.Decimal) {
+          return new AvroDecimalProducer.FixedDecimalProducer((DecimalVector) vector, schema.getFixedSize());
+        } else {
+          return new AvroFixedProducer((FixedSizeBinaryVector) vector, schema.getFixedSize());
+        }
+      case INT:
+        if (logicalType instanceof LogicalTypes.Date) {
+          return new AvroDateProducer((DateDayVector) vector);
+        } else if (logicalType instanceof LogicalTypes.TimeMillis) {
+          return new AvroTimeMillisProducer((TimeMilliVector) vector);
+        } else {
+          return new AvroIntProducer((IntVector) vector);
+        }
+      case BOOLEAN:
+        return new AvroBooleanProducer((BitVector) vector);
+      case LONG:
+        if (logicalType instanceof LogicalTypes.TimeMicros) {
+          return new AvroTimeMicroProducer((TimeMicroVector) vector);
+        } else if (logicalType instanceof LogicalTypes.TimestampMillis) {
+          return new AvroTimestampMillisProducer((TimeStampMilliVector) vector);
+        } else if (logicalType instanceof LogicalTypes.TimestampMicros) {
+          return new AvroTimestampMicroProducer((TimeStampMicroVector) vector);
+        } else {
+          return new AvroLongProducer((BigIntVector) vector);
+        }
+      case FLOAT:
+        return new AvroFloatProducer((Float4Vector) vector);
+      case DOUBLE:
+        return new AvroDoubleProducer((Float8Vector) vector);
+      case BYTES:
+        if (logicalType instanceof LogicalTypes.Decimal) {
+          return new AvroDecimalProducer.BytesDecimalProducer((DecimalVector) vector);
+        } else {
+          return new AvroBytesProducer((VarBinaryVector) vector);
+        }
+      case NULL:
+        return new AvroNullProducer((NullVector) vector);
+      default:
+        throw new UnsupportedOperationException("Unsupported vector type: " + type.getName());
+    }
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroArrayProducer.java

@@ -0,0 +1,58 @@
+/*
+ * 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.arrow.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer which produces array type values to an Avro encoder.
+ * Writes the data from a {@link ListVector}.
+ */
+public class AvroArrayProducer extends BaseAvroProducer<ListVector> {
+
+  private final BaseAvroProducer<?> delegate;
+
+  /** Instantiate an ArrayProducer. */
+  public AvroArrayProducer(ListVector vector, BaseAvroProducer<?> delegate) {
+    super(vector);
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+
+    encoder.writeArrayStart();
+    encoder.setItemCount(vector.getValueCount());
+
+    for (int i = 0; i < vector.getValueCount(); i++) {
+      encoder.startItem();
+      delegate.produce(encoder);
+    }
+
+    encoder.writeArrayEnd();
+    currentIndex++;
+  }
+
+  @Override
+  public boolean resetValueVector(ListVector vector) {
+    this.delegate.resetValueVector(vector.getDataVector());
+    return super.resetValueVector(vector);
+  }
+}