You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@avro.apache.org by th...@apache.org on 2019/05/01 19:11:10 UTC
[avro] branch fast-decoder-thiru created (now faf150c)
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a change to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git.
at faf150c Fixed documentation
This branch includes the following new commits:
new 0ec4b0c Small changes to Resolver to support Advancer
new c602938 First draft of advancer code.
new 345f6b3 Dealt with record field-skipping problem. Also fixed a few errors missed earlier.
new 56410ba Moved Advancer to org.apache.avro.generic package
new a22c22b Fixups from previous two commits.
new cc71ce4 More tweaks
new e23abae Adding alternative to GenericDatumeReader which uses Advancers.
new 32999b8 Changes for what is needed for arrays and maps.
new 3fe8df6 Advancer and GenericDatumeReader2 now compile
new ab9601d Will be handy to have schemas available during decoding.
new 5e50f64 Fleshed out rest of cases in GenericReader2 (but do not handle logical types).
new cefbf86 formatting changes
new e1e9a53 Resolved checkstyle issues
new faf150c Fixed documentation
The 14 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[avro] 02/14: First draft of advancer code.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit c602938347a6e1ba96da11ff7c3e08f8a7eb05ab
Author: rstata <rs...@yahoo.com>
AuthorDate: Mon Apr 29 23:26:46 2019 -0700
First draft of advancer code.
---
.../java/org/apache/avro/specific/Advancer.java | 757 +++++++++++++++++++++
1 file changed, 757 insertions(+)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
new file mode 100644
index 0000000..75d2615
--- /dev/null
+++ b/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
@@ -0,0 +1,757 @@
+/*
+ * 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.avro.specific;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.avro.AvroTypeException;
+import org.apache.avro.Resolver;
+import org.apache.avro.Schema;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.util.Utf8;
+
+
+/** An "Advancer" is a tree of objects that apply resolution logic
+ * while reading values out of a {@link Decoder}.
+ *
+ * An Advancer tree is created by calling {@link advancerFor} on a
+ * {@link Resolver.Action} object. The resulting tree mimics the
+ * reader schema of that Action object.
+ *
+ * A decoder for that reader schema is meant to traverse the schema
+ * in a depth-first fashion. When it hits a leaf of type
+ * <code>Xyz</code>, it should call corresponding
+ * <code>nextXyx</code> on the Advancer. For example, if the reader
+ * hits a lead indicating that an integer should be read, it should
+ * call {@link nextInt}, as in <code>a.nextInt(in)</code>, where
+ * <code>a</code> is the advancer being traversed, and
+ * <code>in</code> is the Decoder being read from.
+ *
+ * When traversing an Array or Map in the reader schema, the decoder
+ * should call {@link getElementAdvancer} to retrieve the advancer
+ * object for the contained element-schema, value-schema, or non-null
+ * schema respectively. ({@link next} cannot be called on {@link
+ * Advancer.Record} objects -- decoders must decode them field by
+ * field.)
+ *
+ * For unions, the decoder should call {@link nextIndex} to fetch the
+ * branch and then {@link getBranchAdvancer} to get the advancer of
+ * that branch. (Calling {@link next} on a union will read the
+ * index, pick the right advancer based on the index, and then read
+ * and return the actual value.)
+ *
+ * Traversing an record is more involved. The decoder should call
+ * {@link getRecordAdvancer} and proceed as described in the
+ * documentation for {@link Advancer.Record}. ({@link next} cannot
+ * be called on {@link Advancer.Record} objects -- decoders must
+ * decode them field by field.)
+ **/
+abstract class Advancer {
+ protected Exception exception() {
+ throw new UnsupportedOperationException();
+ }
+
+ //// API methods of Advancer. Used by decoding methods to
+ //// read values out of Decoder, applying resolution logic
+ //// in the process. In the base class, these do throw
+ //// a not-supported exception. Specific subclasses implement
+ //// certain ones, e.g., IntFast (the Advancer used when
+ //// an integer is read with no promotion) overrides just
+ //// readInt.
+
+ public Object next(Decoder in) throws IOException { exception(); }
+ public Object nextNull(Decoder in) throws IOException { exception(); }
+ public boolean nextBoolean(Decoder in) throws IOException { exception(); }
+ public int nextInt(Decoder in) throws IOException { exception(); }
+ public long nextLong(Decoder in) throws IOException { exception(); }
+ public float nextFloat(Decoder in) throws IOException { exception(); }
+ public double nextDouble(Decoder in) throws IOException { exception(); }
+ public int nextEnum(Decoder in) throws IOException { exception(); }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException { exception(); }
+ public String nextString(Decoder in) throws IOException { exception(); }
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException { exception(); }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException {
+ exception();
+ }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes) throws IOException {
+ return nextFixed(in, bytes, 0, bytes.length);
+ }
+
+ /** Access to contained advancer (for Array and Map types). */
+ public Advancer getElementAdvancer(Decoder in) throws IOException {
+ exception();
+ }
+
+ /** Get index for a union. */
+ public int nextIndex(Decoder in) throws IOException { exception(); }
+
+ /** Access to contained advancer for unions. You must call {@link
+ * nextIndex} before calling this method. */
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
+ exception();
+ }
+
+ /** Access to contained advancer (for Array, Map, and Union types). */
+ public Record getRecordAdvancer(Decoder in) throws IOException {
+ exception();
+ }
+
+
+ ////// Here's the builder for Advancer trees. The subclasses used by
+ ////// this implementation are found below.
+
+ /** Build an {@link Advancer} tree that for a given {@link
+ * Resolver.Action} tree. */
+ public static Advancer from(Resolver.Action a) {
+ switch (a.type) {
+ case DO_NOTHING:
+ switch (a.reader.getType()) {
+ case NULL: return NullFast.instance;
+ case BOOLEAN: return BooleanFast.instance;
+ case INT: return IntFast.instance;
+ case LONG: return LongFast.instance;
+ case FLOAT: return FloatFast.instance;
+ case DOUBLE: return DoubleFast.instance;
+ case STRING: return StringFast.instance;
+ case BYTES: return BytesFast.instance;
+ case FIXED: return new FixedFast(a.writer.getFixedSize());
+ default:
+ throw new IllegalArgumentException("Unexpected schema for DoNothing:" + a.reader);
+ }
+ case PROMOTE:
+ switch (((Resolver.Promote)a).promotion) {
+ case INT2LONG: return LongFromInt.instance;
+ case INT2FLOAT: return FloatFromInt.instance;
+ case INT2DOUBLE: return DoubleFromInt.instance;
+ case LONG2FLOAT: return FloatFromLong.instance;
+ case LONG2DOUBLE: return DoubleFromLong.instance;
+ case FLOAT2DOUBLE: return DoubleFromFloat.instance;
+ case STRING2BYTES: return BytesFromString.instance;
+ case BYTES2STRING: return StringFromBytes.instance;
+ default:
+ throw new IllegalArgumentException("Unexpected promotion:" + a);
+ }
+ case ENUM:
+ Resolver.EnumAdjust e = (Resolver.EnumAdjust)a;
+ if (e.noAdjustmentsNeeded) return EnumFast.instance;
+ else return new EnumWithAdjustments(e.adjustments);
+
+ case CONTAINER:
+ return new Container(Advancer.from(((Resolver.Container)a).elementAction));
+
+ case RECORD:
+ return Advancer.Record.from((Resolver.RecordAdjust)a);
+
+ case WRITER_UNION:
+ Resolver.WriterUnion wu = (Resolver.WriterUnion)a;
+ Advancer[] branches = new Advancer[wu.actions.length];
+ for (int i = 0; i < branches.length; i++)
+ branches[i] = Advancer.from(wu.actions[i]);
+ if (wu.unionEquiv) return new EquivUnion(branches);
+ return new WriterUnion(branches);
+
+ case READER_UNION:
+ Resolver.ReaderUnion ru = (Resolver.ReaderUnion)a;
+ return new ReaderUnion(ru.firstMatch, Advancer.from(ru.actualAction));
+
+ case ERROR:
+ throw new AvroTypeException(a.toString());
+ case SKIP:
+ throw new RuntimeException("Internal error. Skip should've been consumed.");
+ default:
+ throw new IllegalArgumentException("Unknown action:" + a);
+ }
+ }
+
+ private static Schema[] collectSkips(Resolver.Action[] actions, int start) {
+ Schema[] result = EMPTY_SCHEMA_ARRAY;
+ int j = start;
+ while (j < actions.length && actions[j].type == Resolver.Action.Type.SKIP)
+ j++;
+ if (start < j) {
+ result = new Schema[j - start];
+ for (int k = 0; k < (j - start); k++)
+ result[k] = actions[start + k].writer;
+ }
+ return result;
+ }
+ private static final Schema[] EMPTY_SCHEMA_ARRAY = new Schema[0];
+
+ ////// Subclasses of Advancer -- real work is done in these
+
+ /** All methods of <code>this</code> throw {@link
+ * AvroTypeException} with appropriate message. Used for
+ * throwing resolution errors in a lazy fashion (i.e., as actual
+ * data causes the error to manifest). */
+ private static class Error extends Advancer {
+ String msg;
+ public Error(String msg) { this.msg = msg; }
+ protected Exception exception() {
+ throw new AvroTypeException(msg);
+ }
+ }
+
+ /** Used for Array, Map, and Union. In case of Union, since we only
+ * support "nullable" unions (ie, two-branch unions in which one
+ * branch is null), the element advancer is for the non-null branch
+ * of the union. */
+ private static class Container extends Advancer {
+ private final Advancer elementAdvancer;
+ public Advancer getElementAdvancer(Decoder in) { return elementAdvancer; }
+ }
+
+ //// The following set of subclasses are for when there is no
+ //// resolution logic to be applied. All that needs to be done
+ //// is call the corresponding method on the Decoder.
+
+ private static class NullFast extends Advancer {
+ public static final NullFast instance = new NullFast();
+ private NullFast() { }
+ public Object nextNull(Decoder in) throws IOException {
+ in.readNull();
+ return null;
+ }
+ public Object next(Decoder in) throws IOException { return nextNull(in); }
+ }
+
+ private static class BooleanFast extends Advancer {
+ public static final BooleanFast instance = new BooleanFast();
+ private BooleanFast() { }
+ public boolean nextBoolean(Decoder in) throws IOException {
+ return in.readBoolean();
+ }
+ public Object next(Decoder in) throws IOException { return nextBoolean(in); }
+ }
+
+ private static class IntFast extends Advancer {
+ public static final IntFast instance = new IntFast();
+ private IntFast() { }
+ public int nextInt(Decoder in) throws IOException {
+ return in.readInt();
+ }
+ public Object next(Decoder in) throws IOException { return nextInt(in); }
+ }
+
+ private static class LongFast extends Advancer {
+ public static final LongFast instance = new LongFast();
+ private LongFast() { }
+ public long nextLong(Decoder in) throws IOException {
+ return in.readLong();
+ }
+ public Object next(Decoder in) throws IOException { return nextLong(in); }
+ }
+
+ private static class FloatFast extends Advancer {
+ public static final FloatFast instance = new FloatFast();
+ private FloatFast() { }
+ public float nextFloat(Decoder in) throws IOException {
+ return in.readFloat();
+ }
+ public Object next(Decoder in) throws IOException { return nextFloat(in); }
+ }
+
+ private static class DoubleFast extends Advancer {
+ public static final DoubleFast instance = new DoubleFast();
+ private DoubleFast() { }
+ public double nextDouble(Decoder in) throws IOException {
+ return in.readDouble();
+ }
+ public Object next(Decoder in) throws IOException { return nextDouble(in); }
+ }
+
+ private static class StringFast extends Advancer {
+ public static final StringFast instance = new StringFast();
+ private StringFast() { }
+ public String nextString(Decoder in) throws IOException { return in.readString(); }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ return in.readString(old);
+ }
+ public Object next(Decoder in) throws IOException { return nextString(in); }
+ }
+
+ private static class BytesFast extends Advancer {
+ public static final BytesFast instance = new BytesFast();
+ private BytesFast() { }
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ return in.readBytes(old);
+ }
+ public Object next(Decoder in) throws IOException { return nextBytes(in, null); }
+ }
+
+ private static class FixedFast extends Advancer {
+ private final int len;
+ private FixedFast(int len) { this.len = len; }
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
+ in.readFixed(bytes, start, len);
+ return bytes;
+ }
+ public Object next(Decoder in) throws IOException {
+ byte[] result = new byte[len];
+ nextFixed(in, new byte[len]);
+ return result;
+ }
+ }
+
+ private static class EnumFast extends Advancer {
+ public static final EnumFast instance = new EnumFast();
+ private EnumFast() { }
+ public int nextEnum(Decoder in) throws IOException { return in.readEnum(); }
+ public Object next(Decoder in) throws IOException { return nextEnum(in); }
+ }
+
+ //// The following set of subclasses apply promotion logic
+ //// to the underlying value read.
+
+ private static class LongFromInt extends Advancer {
+ public static final LongFromInt instance = new LongFromInt();
+ private LongFromInt() { }
+ public long nextLong(Decoder in) throws IOException {
+ return (long) in.readInt();
+ }
+ public Object next(Decoder in) throws IOException { return nextLong(in); }
+ }
+
+ private static class FloatFromInt extends Advancer {
+ public static final FloatFromInt instance = new FloatFromInt();
+ private FloatFromInt() { }
+ public float nextFloat(Decoder in) throws IOException {
+ return (float) in.readInt();
+ }
+ public Object next(Decoder in) throws IOException { return nextFloat(in); }
+ }
+
+ private static class FloatFromLong extends Advancer {
+ public static final FloatFromLong instance = new FloatFromLong();
+ private FloatFromLong() { }
+ public float nextFloat(Decoder in) throws IOException {
+ return (long) in.readLong();
+ }
+ public Object next(Decoder in) throws IOException { return nextFloat(in); }
+ }
+
+ private static class DoubleFromInt extends Advancer {
+ public static final DoubleFromInt instance = new DoubleFromInt();
+ private DoubleFromInt() { }
+ public double nextDouble(Decoder in) throws IOException {
+ return (double) in.readInt();
+ }
+ public Object next(Decoder in) throws IOException { return nextDouble(in); }
+ }
+
+ private static class DoubleFromLong extends Advancer {
+ public static final DoubleFromLong instance = new DoubleFromLong();
+ private DoubleFromLong() { }
+ public double nextDouble(Decoder in) throws IOException {
+ return (double) in.readLong();
+ }
+ public Object next(Decoder in) throws IOException { return nextDouble(in); }
+ }
+
+ private static class DoubleFromFloat extends Advancer {
+ public static final DoubleFromFloat instance = new DoubleFromFloat();
+ private DoubleFromFloat() { }
+ public double nextDouble(Decoder in) throws IOException {
+ return (double) in.readFloat();
+ }
+ public Object next(Decoder in) throws IOException { return nextDouble(in); }
+ }
+
+ private static class BytesFromString extends Advancer {
+ public static final BytesFromString instance = new BytesFromString();
+ private BytesFromString() { }
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ Utf8 s = in.readString(null);
+ return ByteBuffer.wrap(s.getBytes(), 0, s.getByteLength());
+ }
+ public Object next(Decoder in) throws IOException { return nextBytes(in, null); }
+ }
+
+ private static class StringFromBytes extends Advancer {
+ public static final StringFromBytes instance = new StringFromBytes();
+ private StringFromBytes() { }
+ public String nextString(Decoder in) throws IOException {
+ return new String(in.readBytes(null).array(), StandardCharsets.UTF_8);
+ }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ return new Utf8(in.readBytes(null).array());
+ }
+ public Object next(Decoder in) throws IOException { return nextString(in); }
+ }
+
+
+ //// This last set of advancers are used when more sophisticated
+ //// adjustmentds are needed
+
+ private static class EnumWithAdjustments extends Advancer {
+ private final int[] adjustments;
+ public EnumWithAdjustments(int[] adjustments) {
+ this.adjustments = adjustments;
+ }
+ public int nextEnum(Decoder in) throws IOException {
+ return adjustments[in.readInt()];
+ }
+ public Object next(Decoder in) throws IOException { return nextEnum(in); }
+ }
+
+ /** In this case, the writer has a union by the reader doesn't, so we
+ * consume the tag ourself and call the corresponding advancer. */
+ private static class WriterUnion extends Advancer {
+ private Advancer[] branches;
+ public WriterUnion(Advancer[] branches) { this.branches = branches; }
+
+ private final Advancer b(Decoder in) throws IOException
+ { return branches[in.readIndex()]; }
+
+ public Object next(Decoder in) throws IOException { return b(in).next(in); }
+ public Object nextNull(Decoder in) throws IOException { return b(in).nextNull(in); }
+ public boolean nextBoolean(Decoder in) throws IOException { return b(in).nextBoolean(in); }
+ public int nextInt(Decoder in) throws IOException { return b(in).nextInt(in); }
+ public long nextLong(Decoder in) throws IOException { return b(in).nextLong(in); }
+ public float nextFloat(Decoder in) throws IOException { return b(in).nextFloat(in); }
+ public double nextDouble(Decoder in) throws IOException { return b(in).nextDouble(in); }
+ public int nextEnum(Decoder in) throws IOException { return b(in).nextEnum(in); }
+ public String nextString(Decoder in) throws IOException { return b(in).nextString(in); }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException
+ { return b(in).nextString(in, old); }
+
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException
+ { return b(in).nextBytes(in, old); }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException
+ { return b(in).nextFixed(in, bytes, start, length); }
+
+ public Advancer getElementAdvancer(Decoder in) throws IOException
+ { return b(in).getElementAdvancer(in); }
+
+ public int nextIndex(Decoder in) throws IOException { return b(in).nextIndex(in); }
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
+ { return b(in).getBranchAdvancer(in, branch); }
+
+ public Record getRecordAdvancer(Decoder in) throws IOException
+ { return b(in).getRecordAdvancer(in); }
+ }
+
+ /** In this case, reader and writer have the same union, so let the decoder
+ * consume it as a regular union. */
+ private static class EquivUnion extends Advancer {
+ private final Advancer[] branches;
+ public EquivUnion(Advancer[] branches) { this.branches = branches; }
+
+ public int nextIndex(Decoder in) throws IOException { return in.readIndex(); }
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
+ return branches[branch];
+ }
+ public Object next(Decoder in) throws IOException {
+ return branches[in.readIndex()].next(in);
+ }
+ }
+
+ private static class ReaderUnion extends Advancer {
+ private int branch;
+ private Advancer advancer;
+ public ReaderUnion(int b, Advancer a) { branch = b; advancer = a; }
+ public int nextIndex(Decoder in) { return branch; }
+ public Advancer getBranchAdvancer(Decoder in, int b) {
+ if (b != this.branch)
+ throw new IllegalArgumentException("Branch much be " + branch + ", got " + b);
+ return advancer;
+ }
+ public Object next(Decoder in) throws IOException {
+ return advancer.next(in);
+ }
+ }
+
+
+
+
+ //// Records are particularly intricate because we may have to skip
+ //// fields, read fields out of order, and use default values.
+
+ /** Advancer for records. The {@link advancer} array contains an
+ * advancer for each field, ordered according writer (which
+ * determines the order in which data must be read). The {@link
+ * readerOrder} array tells you how those advancers line up with the
+ * reader's fields. Thus, the following is how to read a record:
+ * <pre>
+ * for (int i = 0; i < a.advancers.length; i++)
+ * dataum.set(a.readerOrder[i], a.advancers[i].next());
+ * </pre>
+ * As a convenience, {@link inOrder} is set to true iff the reader
+ * and writer order agrees (i.e., iff <code>readerOrder[i] ==
+ * i</code> for all i). Generated code can use this to optimize this
+ * common case. */
+ public static class Record extends Advancer {
+ public final Advancer[] advancers;
+ public final int[] readerOrder;
+ public final boolean inOrder;
+
+ private Record(Advancer[] advancers, int[] readerOrder, boolean inOrder) {
+ this.advancers = advancers;
+ this.readerOrder = readerOrder;
+ this.inOrder = inOrder;
+ }
+
+ public Record getRecordAdvancer(Decoder in) { return this; }
+
+ protected static Advancer from(Resolver.RecordAdjust ra) {
+ /** Two cases: reader + writer agree on order, vs disagree. */
+ /** This is the complicated case, since skipping is involved. */
+ /** Special subclasses of Advance will encapsulate skipping. */
+
+ // Compute the "readerOrder" argument to Advancer.Record constructor
+ int[] readOrder = new int[ra.readerOrder.length];
+ for (int i = 0; i < readOrder.length; i++) readOrder[i] = ra.readerOrder[i].pos();
+
+ // Compute the "advancers" argument to Advancer.Record constructor
+ Advancer[] fieldAdvs = new Advancer[readOrder.length];
+
+ int i = 0; // Index into ra.fieldActions
+ int rf = 0; // Index into readOrder
+ int nrf = 0; // Index into fieldAdvs
+
+ // Deal with any leading fields to be skipped
+ Schema[] firstSkips = collectSkips(ra.fieldActions, i);
+ if (firstSkips.length != 0) i += firstSkips.length;
+ else firstSkips = null;
+
+ // Deal with fields to be read
+ for ( ; i < ra.fieldActions.length; nrf++, rf++) {
+ Advancer fieldAdv = Advancer.from(ra.fieldActions[i]);
+ i++;
+ Schema[] toSkip = collectSkips(ra.fieldActions, i);
+ if (toSkip.length != 0) {
+ fieldAdv = new RecordField(fieldAdv, toSkip);
+ i += toSkip.length;
+ }
+ if (firstSkips != null) {
+ fieldAdv = new RecordFieldWithBefore(firstSkips, fieldAdv);
+ firstSkips = null;
+ }
+ fieldAdvs[nrf] = fieldAdv;
+ }
+
+ // If reader and writer orders agree, sort fieldAdvs by reader
+ // order (i.e., move defaults into the correct place), to allow
+ // decoders to have an optimized path for the common case of a
+ // record's field order not changing.
+ boolean inOrder = true;
+ for (int k = 0; k < ra.firstDefault-1; k++)
+ if (readOrder[k] > readOrder[k+1]) inOrder = false;
+ if (inOrder) {
+ Advancer[] newAdvancers = new Advancer[fieldAdvs.length];
+ for (int k = 0, rf2 = 0, df = ra.firstDefault; k < readOrder.length; k++) {
+ if (rf2 < df) newAdvancers[k] = fieldAdvs[rf2++];
+ else newAdvancers[k] = fieldAdvs[df++];
+ readOrder[k] = k;
+ }
+ newAdvancers = fieldAdvs;
+ }
+
+ return new Record(fieldAdvs, readOrder, inOrder);
+ }
+ }
+
+ private static class RecordField extends Advancer {
+ private final Advancer field;
+ private final Schema[] after;
+ public RecordField(Advancer field, Schema[] after) {
+ this.field = field;
+ this.after = after;
+ }
+
+ public Object next(Decoder in) throws IOException
+ { Object r = field.next(in); ignore(after, in); return r; }
+
+ public Object nextNull(Decoder in) throws IOException
+ { field.nextNull(in); ignore(after, in); return null; }
+
+ public boolean nextBoolean(Decoder in) throws IOException
+ { boolean r = field.nextBoolean(in); ignore(after, in); return r; }
+
+ public int nextInt(Decoder in) throws IOException
+ { int r = field.nextInt(in); ignore(after, in); return r; }
+
+ public long nextLong(Decoder in) throws IOException
+ { long r = field.nextLong(in); ignore(after, in); return r; }
+
+ public float nextFloat(Decoder in) throws IOException
+ { float r = field.nextFloat(in); ignore(after, in); return r; }
+
+ public double nextDouble(Decoder in) throws IOException
+ { double r = field.nextDouble(in); ignore(after, in); return r; }
+
+ public int nextEnum(Decoder in) throws IOException
+ { int r = field.nextEnum(in); ignore(after, in); return r; }
+
+ public String nextString(Decoder in) throws IOException
+ { String r = field.nextString(in); ignore(after, in); return r; }
+
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ Utf8 r = field.nextString(in,old);
+ ignore(after, in);
+ return r;
+ }
+
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ ByteBuffer r = field.nextBytes(in,old);
+ ignore(after, in);
+ return r;
+ }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
+ byte[] r = field.nextFixed(in, bytes, start, len);
+ ignore(after, in);
+ return r;
+ }
+
+ // TODO: THIS DOESN'T WORK!!
+ public Advancer getElementAdvancer(Decoder in) throws IOException
+ { Advancer r = field.getElementAdvancer(in); ignore(after, in); return r; }
+
+ // TODO: THIS DOESN'T WORK!!
+ public int nextIndex(Decoder in) throws IOException
+ { int r = field.nextIndex(in); ignore(after, in); return r; }
+
+ // TODO: THIS DOESN'T WORK!!
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
+ { Advancer r = field.getBranchAdvancer(in, branch); ignore(after, in); return r; }
+
+ // TODO: THIS DOESN'T WORK!!
+ public Record getRecordAdvancer(Decoder in) throws IOException
+ { Record r = field.getRecordAdvancer(in); ignore(after, in); return r; }
+ }
+
+ private static class RecordFieldWithBefore extends Advancer {
+ private final Schema[] before;
+ private final Advancer field;
+ public RecordFieldWithBefore(Schema[] before, Advancer field) {
+ this.before = before;
+ this.field = field;
+ }
+
+ public Object next(Decoder in) throws IOException
+ { ignore(before, in); return field.next(in); }
+
+ public Object nextNull(Decoder in) throws IOException
+ { ignore(before, in); return field.nextNull(in); }
+
+ public boolean nextBoolean(Decoder in) throws IOException
+ { ignore(before, in); return field.nextBoolean(in); }
+
+ public int nextInt(Decoder in) throws IOException
+ { ignore(before, in); return field.nextInt(in); }
+
+ public long nextLong(Decoder in) throws IOException
+ { ignore(before, in); return field.nextLong(in); }
+
+ public float nextFloat(Decoder in) throws IOException
+ { ignore(before, in); return field.nextFloat(in); }
+
+ public double nextDouble(Decoder in) throws IOException
+ { ignore(before, in); return field.nextDouble(in); }
+
+ public int nextEnum(Decoder in) throws IOException
+ { ignore(before, in); return field.nextEnum(in); }
+
+ public String nextString(Decoder in) throws IOException
+ { ignore(before, in); return field.nextString(in); }
+
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException
+ { ignore(before, in); return field.nextString(in, old); }
+
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException
+ { ignore(before, in); return field.nextBytes(in, old); }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException
+ { ignore(before, in); return field.nextFixed(in, bytes, start, len); }
+
+ public Advancer getElementAdvancer(Decoder in) throws IOException
+ { ignore(before, in); return field.getElementAdvancer(in); }
+
+ public int nextIndex(Decoder in) throws IOException
+ { ignore(before, in); return field.nextIndex(in); }
+
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
+ { ignore(before, in); return field.getBranchAdvancer(in, branch); }
+
+ public Record getRecordAdvancer(Decoder in) throws IOException
+ { ignore(before, in); return field.getRecordAdvancer(in); }
+ }
+
+
+ private static class Default extends Advancer {
+ protected final Object val;
+ private Default(Object val) { this.val = val; }
+
+ public Object next(Decoder in) { return val; }
+ public Object nextNull(Decoder in) { return val; }
+ public boolean nextBoolean(Decoder in) { return (Boolean) val; }
+ public int nextInt(Decoder in) { return (Integer) val; }
+ public long nextLong(Decoder in) { return (Long) val; }
+ public float nextFloat(Decoder in) { return (Float) val; }
+ public double nextDouble(Decoder in) { return (Double) val; }
+ public int nextEnum(Decoder in) { return (Integer) val; }
+
+ // TODO -- finish for the rest of the types
+ }
+
+ private static void ignore(Schema[] toIgnore, Decoder in) throws IOException {
+ for (Schema s: toIgnore) skip(s, in);
+ }
+
+ // Probably belongs someplace else, although Decoder doesn't reference
+ // Schema, and Schema doesn't reference Decoder, and I'd hate to create
+ // new dependencies...
+ public static void skip(Schema s, Decoder in) throws IOException {
+ switch (s.getType()) {
+ case NULL: in.readNull(); break;
+ case BOOLEAN: in.readBoolean(); break;
+ case INT: in.readInt(); break;
+ case LONG: in.readLong(); break;
+ case FLOAT: in.readFloat(); break;
+ case DOUBLE: in.readDouble(); break;
+ case STRING: in.skipString(); break;
+ case BYTES: in.skipBytes(); break;
+ case FIXED: in.skipFixed(s.getFixedSize()); break;
+ case ENUM: in.readEnum(); break;
+ case UNION: skip(s.getTypes().get(in.readInt()), in); break;
+ case RECORD:
+ for (Schema.Field f: s.getFields())
+ skip(f.schema(), in);
+ break;
+ case ARRAY:
+ for (long i = in.skipArray(); i != 0; i = in.skipArray())
+ for (long j = 0; j < i; j++)
+ skip(s.getElementType(), in);
+ break;
+ case MAP:
+ for (long k = in.skipArray(); k != 0; k = in.skipArray())
+ for (long l = 0; l < k; l++) {
+ in.skipString(); // Key
+ skip(s.getValueType(), in);
+ }
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown type for schema: " + s);
+ }
+ }
+}
[avro] 04/14: Moved Advancer to org.apache.avro.generic package
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 56410ba64905509a62a4a59b23605436be1cc971
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 00:21:47 2019 -0700
Moved Advancer to org.apache.avro.generic package
---
.../src/main/java/org/apache/avro/{specific => generic}/Advancer.java | 0
1 file changed, 0 insertions(+), 0 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
similarity index 100%
rename from lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
rename to lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
[avro] 10/14: Will be handy to have schemas available during
decoding.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit ab9601dbfeff198d41bc9fb5fe419d22ec8d14e7
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 13:13:55 2019 -0700
Will be handy to have schemas available during decoding.
---
.../java/org/apache/avro/generic/Advancer.java | 111 +++++++++++++--------
1 file changed, 69 insertions(+), 42 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 8f2b010..d80e812 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -75,6 +75,9 @@ abstract class Advancer {
//// an integer is read with no promotion) overrides just
//// readInt.
+ public final Schema writer, reader;
+ protected Advancer(Schema w, Schema r) { this.writer = w; this.reader = r; }
+
public Object next(Decoder in) throws IOException { exception(); return null; }
public Object nextNull(Decoder in) throws IOException { exception(); return null; }
public boolean nextBoolean(Decoder in) throws IOException { exception(); return false; }
@@ -140,7 +143,7 @@ abstract class Advancer {
case DOUBLE: return DoubleFast.instance;
case STRING: return StringFast.instance;
case BYTES: return BytesFast.instance;
- case FIXED: return new FixedFast(a.writer.getFixedSize());
+ case FIXED: return new FixedFast(a.writer, a.reader);
default:
throw new IllegalArgumentException("Unexpected schema for DoNothing:" + a.reader);
}
@@ -159,13 +162,14 @@ abstract class Advancer {
}
case ENUM:
Resolver.EnumAdjust e = (Resolver.EnumAdjust)a;
- if (e.noAdjustmentsNeeded) return EnumFast.instance;
- else return new EnumWithAdjustments(e.adjustments);
+ if (e.noAdjustmentsNeeded) return new EnumFast(a.writer, a.reader);
+ else return new EnumWithAdjustments(a.writer, a.reader, e.adjustments);
case CONTAINER:
Advancer ea = Advancer.from(((Resolver.Container)a).elementAction);
- if (a.writer.getType() == Schema.Type.ARRAY) return new ArrayContainer(ea);
- else return new MapContainer(ea);
+ if (a.writer.getType() == Schema.Type.ARRAY)
+ return new ArrayContainer(a.writer, a.reader, ea);
+ else return new MapContainer(a.writer, a.reader, ea);
case RECORD:
return Advancer.Record.from((Resolver.RecordAdjust)a);
@@ -175,15 +179,16 @@ abstract class Advancer {
Advancer[] branches = new Advancer[wu.actions.length];
for (int i = 0; i < branches.length; i++)
branches[i] = Advancer.from(wu.actions[i]);
- if (wu.unionEquiv) return new EquivUnion(branches);
- return new WriterUnion(branches);
+ if (wu.unionEquiv) return new EquivUnion(a.writer, a.reader, branches);
+ return new WriterUnion(a.writer, a.reader, branches);
case READER_UNION:
Resolver.ReaderUnion ru = (Resolver.ReaderUnion)a;
- return new ReaderUnion(ru.firstMatch, Advancer.from(ru.actualAction));
+ return new ReaderUnion(a.writer, a.reader,
+ ru.firstMatch, Advancer.from(ru.actualAction));
case ERROR:
- throw new AvroTypeException(a.toString());
+ return new Error(w,r, a.toString());
case SKIP:
throw new RuntimeException("Internal error. Skip should've been consumed.");
default:
@@ -213,7 +218,7 @@ abstract class Advancer {
* data causes the error to manifest). */
private static class Error extends Advancer {
String msg;
- public Error(String msg) { this.msg = msg; }
+ public Error(Schema w, Schema r, String msg) { super(w,r); this.msg = msg; }
protected Exception exception() {
throw new AvroTypeException(msg);
}
@@ -236,7 +241,7 @@ abstract class Advancer {
* illustrations. */
public abstract static class Container extends Advancer {
private final Advancer elementAdvancer;
- public Container(Advancer elementAdvancer) { this.elementAdvancer = elementAdvancer; }
+ public Container(Schema w, Schema r, Advancer ea) { super(wr); elementAdvancer = ea; }
public Container getContainerAdvancer(Decoder in) { return this; }
public Advancer getElementAdvancer(Decoder in) { return elementAdvancer; }
public abstract long firstChunk(Decoder in) throws IOException;
@@ -244,7 +249,7 @@ abstract class Advancer {
}
private static class ArrayContainer extends Container {
- public ArrayContainer(Advancer ea) { super(ea); }
+ public ArrayContainer(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
public long firstChunk(Decoder in) throws IOException
{ return in.readArrayStart(); }
public long nextChunk(Decoder in) throws IOException
@@ -252,7 +257,7 @@ abstract class Advancer {
}
private static class MapContainer extends Container {
- public MapContainer(Advancer ea) { super(ea); }
+ public MapContainer(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
public long firstChunk(Decoder in) throws IOException
{ return in.readMapStart(); }
public long nextChunk(Decoder in) throws IOException
@@ -265,7 +270,7 @@ abstract class Advancer {
private static class NullFast extends Advancer {
public static final NullFast instance = new NullFast();
- private NullFast() { }
+ private NullFast() { Schema s = Schema.create(Schema.Type.NULL); super(s,s); }
public Object nextNull(Decoder in) throws IOException {
in.readNull();
return null;
@@ -275,7 +280,7 @@ abstract class Advancer {
private static class BooleanFast extends Advancer {
public static final BooleanFast instance = new BooleanFast();
- private BooleanFast() { }
+ private BooleanFast() { Schema s = Schema.create(Schema.Type.BOOLEAN); super(s,s); }
public boolean nextBoolean(Decoder in) throws IOException {
return in.readBoolean();
}
@@ -284,7 +289,7 @@ abstract class Advancer {
private static class IntFast extends Advancer {
public static final IntFast instance = new IntFast();
- private IntFast() { }
+ private IntFast() { Schema s = Schema.create(Schema.Type.INTEGER); super(s,s); }
public int nextInt(Decoder in) throws IOException {
return in.readInt();
}
@@ -293,7 +298,7 @@ abstract class Advancer {
private static class LongFast extends Advancer {
public static final LongFast instance = new LongFast();
- private LongFast() { }
+ private LongFast() { Schema s = Schema.create(Schema.Type.LONG); super(s,s); }
public long nextLong(Decoder in) throws IOException {
return in.readLong();
}
@@ -302,7 +307,7 @@ abstract class Advancer {
private static class FloatFast extends Advancer {
public static final FloatFast instance = new FloatFast();
- private FloatFast() { }
+ private FloatFast() { Schema s = Schema.create(Schema.Type.FLOAT); super(s,s); }
public float nextFloat(Decoder in) throws IOException {
return in.readFloat();
}
@@ -311,7 +316,7 @@ abstract class Advancer {
private static class DoubleFast extends Advancer {
public static final DoubleFast instance = new DoubleFast();
- private DoubleFast() { }
+ private DoubleFast() { Schema s = Schema.create(Schema.Type.DOUBLE); super(s,s); }
public double nextDouble(Decoder in) throws IOException {
return in.readDouble();
}
@@ -320,7 +325,7 @@ abstract class Advancer {
private static class StringFast extends Advancer {
public static final StringFast instance = new StringFast();
- private StringFast() { }
+ private StringFast() { Schema s = Schema.create(Schema.Type.STRING); super(s,s); }
public String nextString(Decoder in) throws IOException { return in.readString(); }
public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
return in.readString(old);
@@ -330,7 +335,7 @@ abstract class Advancer {
private static class BytesFast extends Advancer {
public static final BytesFast instance = new BytesFast();
- private BytesFast() { }
+ private BytesFast() { Schema s = Schema.create(Schema.Type.BYTES); super(s,s); }
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
return in.readBytes(old);
}
@@ -339,7 +344,7 @@ abstract class Advancer {
private static class FixedFast extends Advancer {
private final int len;
- private FixedFast(int len) { this.len = len; }
+ private FixedFast(Schema w, Schema r) { super(w,r); this.len = w.getFixedSize(); }
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
in.readFixed(bytes, start, len);
return bytes;
@@ -352,8 +357,7 @@ abstract class Advancer {
}
private static class EnumFast extends Advancer {
- public static final EnumFast instance = new EnumFast();
- private EnumFast() { }
+ public EnumFast(Schema w, Schema r) { super(w,r); }
public int nextEnum(Decoder in) throws IOException { return in.readEnum(); }
public Object next(Decoder in) throws IOException { return nextEnum(in); }
}
@@ -363,7 +367,9 @@ abstract class Advancer {
private static class LongFromInt extends Advancer {
public static final LongFromInt instance = new LongFromInt();
- private LongFromInt() { }
+ private LongFromInt() {
+ super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.LONG));
+ }
public long nextLong(Decoder in) throws IOException {
return (long) in.readInt();
}
@@ -372,7 +378,9 @@ abstract class Advancer {
private static class FloatFromInt extends Advancer {
public static final FloatFromInt instance = new FloatFromInt();
- private FloatFromInt() { }
+ private FloatFromInt() {
+ super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.FLOAT));
+ }
public float nextFloat(Decoder in) throws IOException {
return (float) in.readInt();
}
@@ -381,7 +389,9 @@ abstract class Advancer {
private static class FloatFromLong extends Advancer {
public static final FloatFromLong instance = new FloatFromLong();
- private FloatFromLong() { }
+ private FloatFromLong() {
+ super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.FLOAT));
+ }
public float nextFloat(Decoder in) throws IOException {
return (long) in.readLong();
}
@@ -390,7 +400,9 @@ abstract class Advancer {
private static class DoubleFromInt extends Advancer {
public static final DoubleFromInt instance = new DoubleFromInt();
- private DoubleFromInt() { }
+ private DoubleFromInt() {
+ super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.DOUBLE));
+ }
public double nextDouble(Decoder in) throws IOException {
return (double) in.readInt();
}
@@ -399,7 +411,9 @@ abstract class Advancer {
private static class DoubleFromLong extends Advancer {
public static final DoubleFromLong instance = new DoubleFromLong();
- private DoubleFromLong() { }
+ private DoubleFromLong() {
+ super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.DOUBLE));
+ }
public double nextDouble(Decoder in) throws IOException {
return (double) in.readLong();
}
@@ -408,7 +422,9 @@ abstract class Advancer {
private static class DoubleFromFloat extends Advancer {
public static final DoubleFromFloat instance = new DoubleFromFloat();
- private DoubleFromFloat() { }
+ private DoubleFromFloat() {
+ super(Schema.create(Schema.Type.FLOAT), Schema.create(Schema.Type.DOUBLE));
+ }
public double nextDouble(Decoder in) throws IOException {
return (double) in.readFloat();
}
@@ -417,7 +433,9 @@ abstract class Advancer {
private static class BytesFromString extends Advancer {
public static final BytesFromString instance = new BytesFromString();
- private BytesFromString() { }
+ private BytesFromString() {
+ super(Schema.create(Schema.Type.STRING), Schema.create(Schema.Type.BYTES));
+ }
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
Utf8 s = in.readString(null);
return ByteBuffer.wrap(s.getBytes(), 0, s.getByteLength());
@@ -427,7 +445,9 @@ abstract class Advancer {
private static class StringFromBytes extends Advancer {
public static final StringFromBytes instance = new StringFromBytes();
- private StringFromBytes() { }
+ private StringFromBytes() {
+ super(Schema.create(Schema.Type.BYTES), Schema.create(Schema.Type.STRING));
+ }
public String nextString(Decoder in) throws IOException {
return new String(in.readBytes(null).array(), StandardCharsets.UTF_8);
}
@@ -443,7 +463,8 @@ abstract class Advancer {
private static class EnumWithAdjustments extends Advancer {
private final int[] adjustments;
- public EnumWithAdjustments(int[] adjustments) {
+ public EnumWithAdjustments(Schema w, Schema r, int[] adjustments) {
+ super(w,r);
this.adjustments = adjustments;
}
public int nextEnum(Decoder in) throws IOException {
@@ -456,7 +477,7 @@ abstract class Advancer {
* consume the tag ourself and call the corresponding advancer. */
private static class WriterUnion extends Advancer {
private Advancer[] branches;
- public WriterUnion(Advancer[] branches) { this.branches = branches; }
+ public WriterUnion(Schema w, Schema r, Advancer[] b) { super(w,r) branches = b; }
private final Advancer b(Decoder in) throws IOException
{ return branches[in.readIndex()]; }
@@ -494,7 +515,7 @@ abstract class Advancer {
* consume it as a regular union. */
private static class EquivUnion extends Advancer {
private final Advancer[] branches;
- public EquivUnion(Advancer[] branches) { this.branches = branches; }
+ public EquivUnion(Schema w, Schema r, Advancer[] b) {super(w,r); branches = b; }
public int nextIndex(Decoder in) throws IOException { return in.readIndex(); }
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
@@ -505,8 +526,11 @@ abstract class Advancer {
private static class ReaderUnion extends Advancer {
private int branch;
private Advancer advancer;
- public ReaderUnion(int b, Advancer a) { branch = b; advancer = a; }
+ public ReaderUnion(Schema w, Schema r, int b, Advancer a)
+ { super(w,r); branch = b; advancer = a; }
+
public int nextIndex(Decoder in) { return branch; }
+
public Advancer getBranchAdvancer(Decoder in, int b) {
if (b != this.branch)
throw new IllegalArgumentException("Branch much be " + branch + ", got " + b);
@@ -544,9 +568,10 @@ abstract class Advancer {
public final Schema.Field[] readerOrder;
public final boolean inOrder;
- private Record(Advancer[] advancers, Schema[] finalSkips,
+ private Record(Schema w, Schema r, Advancer[] advancers, Schema[] finalSkips,
Schema.Field[] readerOrder, boolean inOrder)
{
+ super(w,r);
this.advancers = advancers;
this.finalSkips = finalSkips;
this.readerOrder = readerOrder;
@@ -580,7 +605,8 @@ abstract class Advancer {
Schema[] toSkip = collectSkips(ra.fieldActions, i);
i += toSkip.length;
Advancer fieldAdv = Advancer.from(ra.fieldActions[i++]);
- if (toSkip.length != 0) fieldAdv = new RecordField(toSkip, fieldAdv);
+ if (toSkip.length != 0)
+ fieldAdv = new RecordField(fieldAdv.writer, fieldAdv.reader, toSkip, fieldAdv);
fieldAdvs[nrf] = fieldAdv;
}
@@ -590,7 +616,7 @@ abstract class Advancer {
// Deal with defaults
for (int df = 0; rf < readOrder.length; rf++, df++, nrf++)
- fieldAdvs[nrf] = new Default(ra.defaults[df]);
+ fieldAdvs[nrf] = new Default(ra.readerOrder[df].schema(), ra.defaults[df]);
// If reader and writer orders agree, sort fieldAdvs by reader
// order (i.e., move defaults into the correct place), to allow
@@ -615,14 +641,15 @@ abstract class Advancer {
readOrder = newReadOrder;
}
- return new Record(fieldAdvs, finalSkips, readOrder, inOrder);
+ return new Record(ra.writer, ra.reader, fieldAdvs, finalSkips, readOrder, inOrder);
}
}
private static class RecordField extends Advancer {
private final Schema[] toSkip;
private final Advancer field;
- public RecordField(Schema[] toSkip, Advancer field) {
+ public RecordField(Schema w, Schema r, Schema[] toSkip, Advancer field) {
+ super(w,r);
this.toSkip = toSkip;
this.field = field;
}
@@ -675,7 +702,7 @@ abstract class Advancer {
private static class Default extends Advancer {
protected final Object val;
- private Default(Object val) { this.val = val; }
+ private Default(Schema s, Object v) { super(s,s); val = v; }
public Object next(Decoder in) { return val; }
public Object nextNull(Decoder in) { return val; }
[avro] 14/14: Fixed documentation
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit faf150c3dbe06447a1264c3ad8c5ed3f2e1ee848
Author: Thiruvalluvan M G <th...@startsmartlabs.com>
AuthorDate: Wed May 1 12:10:16 2019 -0700
Fixed documentation
---
.../java/org/apache/avro/generic/Advancer.java | 84 +++++++++-------------
.../apache/avro/generic/GenericDatumReader2.java | 1 -
2 files changed, 32 insertions(+), 53 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 0f48994..1fdac92 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -31,7 +31,7 @@ import org.apache.avro.util.Utf8;
* An "Advancer" is a tree of objects that apply resolution logic while reading
* values out of a {@link Decoder}.
*
- * An Advancer tree is created by calling {@link advancerFor} on a
+ * An Advancer tree is created by calling {@link Advancer#from(Resolver.Action)} on a
* {@link Resolver.Action} object. The resulting tree mimics the reader schema
* of that Action object.
*
@@ -39,30 +39,27 @@ import org.apache.avro.util.Utf8;
* depth-first fashion. When it hits a leaf of type <code>Xyz</code>, it should
* call corresponding <code>nextXyx</code> on the Advancer. For example, if the
* reader hits a lead indicating that an integer should be read, it should call
- * {@link nextInt}, as in <code>a.nextInt(in)</code>, where <code>a</code> is
+ * {@link Advancer#nextInt}, as in <code>a.nextInt(in)</code>, where <code>a</code> is
* the advancer being traversed, and <code>in</code> is the Decoder being read
* from.
*
* When traversing an Array or Map in the reader schema, the decoder should call
- * {@link getElementAdvancer} to retrieve the advancer object for the contained
+ * {@link Advancer.Container#elementAdvancer} to retrieve the advancer object for the contained
* element-schema or value-schema. See the JavaDoc for
- * {@link getElementAdvancer} for instructions on how to decode these types.
+ * {@link Advancer.Container#elementAdvancer} for instructions on how to decode these types.
*
- * For unions, the decoder should call {@link nextIndex} to fetch the branch and
- * then {@link getBranchAdvancer} to get the advancer of that branch. (Calling
- * {@link next} on a union will read the index, pick the right advancer based on
+ * For unions, the decoder should call {@link Advancer#nextIndex} to fetch the branch and
+ * then {@link Advancer#getBranchAdvancer} to get the advancer of that branch. (Calling
+ * {@link Advancer#next} on a union will read the index, pick the right advancer based on
* the index, and then read and return the actual value.)
*
* Traversing records, arrays, and maps is more involved. In the case of an
- * array or map, call {@link getArrayAdvancer} {@link getMapAdvancer} and
+ * array or map, call {@link Advancer#getArrayAdvancer} {@link Advancer#getMapAdvancer} and
* proceed as described in the documentation for {@link Advancer.Container}. For
* records, best to just look at the implementation of
* {@link GenericDatumReader2}.
**/
abstract class Advancer {
- protected Exception exception() {
- throw new UnsupportedOperationException();
- }
//// API methods of Advancer. Used by decoding methods to
//// read values out of Decoder, applying resolution logic
@@ -80,63 +77,51 @@ abstract class Advancer {
}
public Object next(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public Object nextNull(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public boolean nextBoolean(Decoder in) throws IOException {
- exception();
- return false;
+ throw new UnsupportedOperationException();
}
public int nextInt(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
public long nextLong(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
public float nextFloat(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
public double nextDouble(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
public int nextEnum(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public String nextString(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
public byte[] nextFixed(Decoder in, byte[] bytes) throws IOException {
@@ -145,35 +130,30 @@ abstract class Advancer {
/** Get index for a union. */
public int nextIndex(Decoder in) throws IOException {
- exception();
- return 0;
+ throw new UnsupportedOperationException();
}
/**
- * Access to contained advancer for unions. You must call {@link nextIndex}
+ * Access to contained advancer for unions. You must call {@link Advancer#nextIndex}
* before calling this method.
*/
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
/** Access to advancer for array type. */
public Container getArrayAdvancer(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
/** Access to advancer for array type. */
public Map getMapAdvancer(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
/** Access to advancer for record type. */
public Record getRecordAdvancer(Decoder in) throws IOException {
- exception();
- return null;
+ throw new UnsupportedOperationException();
}
////// Here's the builder for Advancer trees. The subclasses used by
@@ -836,9 +816,9 @@ abstract class Advancer {
//// fields, read fields out of order, and use default values.
/**
- * Advancer for records. The {@link advancers} array contains an advancer for
+ * Advancer for records. The {@link Advancer.Record#advancers} array contains an advancer for
* each field, ordered according writer (which determines the order in which
- * data must be read). The {@link readerOrder} array tells you how those
+ * data must be read). The {@link Advancer.Record#readerOrder} array tells you how those
* advancers line up with the reader's fields. Thus, the following is how to
* read a record:
*
@@ -849,10 +829,10 @@ abstract class Advancer {
* a.done(in);
* </pre>
*
- * Note that a decoder <em>must</em> call {@link done} after interpreting all
- * the elemnts in {@link advancers}.
+ * Note that a decoder <em>must</em> call {@link Advancer.Record#done} after interpreting all
+ * the elemnts in {@link Advancer.Record#advancers}.
*
- * As a convenience, {@link inOrder} is set to true iff the reader and writer
+ * As a convenience, {@link Advancer.Record#inOrder} is set to true iff the reader and writer
* order agrees (i.e., iff <code>readerOrder[i].pos() ==
* i</code> for all i). Generated code can use this to optimize this common
* case.
@@ -876,7 +856,7 @@ abstract class Advancer {
return this;
}
- /** Must be called after consuming all elements of {@link advancers}. */
+ /** Must be called after consuming all elements of {@link Advancer.Record#advancers}. */
public void done(Decoder in) throws IOException {
ignore(finalSkips, in);
}
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
index 19c77a8..a3c4591 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -144,4 +144,3 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
throw new UnsupportedOperationException();
}
}
-
[avro] 13/14: Resolved checkstyle issues
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit e1e9a536368a086935faeef6bacd3bc6f5b4940b
Author: Thiruvalluvan M G <th...@startsmartlabs.com>
AuthorDate: Wed May 1 10:25:42 2019 -0700
Resolved checkstyle issues
---
.../src/main/java/org/apache/avro/Resolver.java | 2 +-
.../java/org/apache/avro/generic/Advancer.java | 98 +++++++++++-----------
.../apache/avro/generic/GenericDatumReader2.java | 3 +-
3 files changed, 52 insertions(+), 51 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
index a01dfb9..bba7789 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
@@ -318,7 +318,7 @@ public class Resolver {
return Promotions.STRING2BYTES;
break;
case STRING:
- if (wt == Schema.Type.STRING)
+ if (wt == Schema.Type.BYTES)
return Promotions.BYTES2STRING;
break;
}
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 21bb22a..0f48994 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -189,21 +189,21 @@ abstract class Advancer {
case DO_NOTHING:
switch (a.reader.getType()) {
case NULL:
- return NullFast.instance;
+ return NullFast.INSTANCE;
case BOOLEAN:
- return BooleanFast.instance;
+ return BooleanFast.INSTANCE;
case INT:
- return IntFast.instance;
+ return IntFast.INSTANCE;
case LONG:
- return LongFast.instance;
+ return LongFast.INSTANCE;
case FLOAT:
- return FloatFast.instance;
+ return FloatFast.INSTANCE;
case DOUBLE:
- return DoubleFast.instance;
+ return DoubleFast.INSTANCE;
case STRING:
- return StringFast.instance;
+ return StringFast.INSTANCE;
case BYTES:
- return BytesFast.instance;
+ return BytesFast.INSTANCE;
case FIXED:
return new FixedFast(a.writer, a.reader);
default:
@@ -212,21 +212,21 @@ abstract class Advancer {
case PROMOTE:
switch (((Resolver.Promote) a).promotion) {
case INT2LONG:
- return LongFromInt.instance;
+ return LongFromInt.INSTANCE;
case INT2FLOAT:
- return FloatFromInt.instance;
+ return FloatFromInt.INSTANCE;
case INT2DOUBLE:
- return DoubleFromInt.instance;
+ return DoubleFromInt.INSTANCE;
case LONG2FLOAT:
- return FloatFromLong.instance;
+ return FloatFromLong.INSTANCE;
case LONG2DOUBLE:
- return DoubleFromLong.instance;
+ return DoubleFromLong.INSTANCE;
case FLOAT2DOUBLE:
- return DoubleFromFloat.instance;
+ return DoubleFromFloat.INSTANCE;
case STRING2BYTES:
- return BytesFromString.instance;
+ return BytesFromString.INSTANCE;
case BYTES2STRING:
- return StringFromBytes.instance;
+ return StringFromBytes.INSTANCE;
default:
throw new IllegalArgumentException("Unexpected promotion:" + a);
}
@@ -355,7 +355,7 @@ abstract class Advancer {
* See the implementation of {@link GenericDatumReader2} for more illustrations.
*/
public static class Map extends Container {
- public final Advancer keyAdvancer = StringFast.instance;
+ public final Advancer keyAdvancer = StringFast.INSTANCE;
public Map(Schema w, Schema r, Advancer ea) {
super(w, r, ea);
@@ -375,11 +375,11 @@ abstract class Advancer {
//// is call the corresponding method on the Decoder.
private static class NullFast extends Advancer {
- public static final NullFast instance = new NullFast();
- private static final Schema s = Schema.create(Schema.Type.NULL);
+ public static final NullFast INSTANCE = new NullFast();
+ private static final Schema S = Schema.create(Schema.Type.NULL);
private NullFast() {
- super(s, s);
+ super(S, S);
}
public Object nextNull(Decoder in) throws IOException {
@@ -393,11 +393,11 @@ abstract class Advancer {
}
private static class BooleanFast extends Advancer {
- public static final BooleanFast instance = new BooleanFast();
- private static final Schema s = Schema.create(Schema.Type.BOOLEAN);
+ public static final BooleanFast INSTANCE = new BooleanFast();
+ private static final Schema S = Schema.create(Schema.Type.BOOLEAN);
private BooleanFast() {
- super(s, s);
+ super(S, S);
}
public boolean nextBoolean(Decoder in) throws IOException {
@@ -410,11 +410,11 @@ abstract class Advancer {
}
private static class IntFast extends Advancer {
- public static final IntFast instance = new IntFast();
- private static final Schema s = Schema.create(Schema.Type.INT);
+ public static final IntFast INSTANCE = new IntFast();
+ private static final Schema S = Schema.create(Schema.Type.INT);
private IntFast() {
- super(s, s);
+ super(S, S);
}
public int nextInt(Decoder in) throws IOException {
@@ -427,11 +427,11 @@ abstract class Advancer {
}
private static class LongFast extends Advancer {
- public static final LongFast instance = new LongFast();
- private static final Schema s = Schema.create(Schema.Type.LONG);
+ public static final LongFast INSTANCE = new LongFast();
+ private static final Schema S = Schema.create(Schema.Type.LONG);
private LongFast() {
- super(s, s);
+ super(S, S);
}
public long nextLong(Decoder in) throws IOException {
@@ -444,11 +444,11 @@ abstract class Advancer {
}
private static class FloatFast extends Advancer {
- public static final FloatFast instance = new FloatFast();
- private static final Schema s = Schema.create(Schema.Type.FLOAT);
+ public static final FloatFast INSTANCE = new FloatFast();
+ private static final Schema S = Schema.create(Schema.Type.FLOAT);
private FloatFast() {
- super(s, s);
+ super(S, S);
}
public float nextFloat(Decoder in) throws IOException {
@@ -461,11 +461,11 @@ abstract class Advancer {
}
private static class DoubleFast extends Advancer {
- public static final DoubleFast instance = new DoubleFast();
- private static final Schema s = Schema.create(Schema.Type.DOUBLE);
+ public static final DoubleFast INSTANCE = new DoubleFast();
+ private static final Schema S = Schema.create(Schema.Type.DOUBLE);
private DoubleFast() {
- super(s, s);
+ super(S, S);
}
public double nextDouble(Decoder in) throws IOException {
@@ -478,11 +478,11 @@ abstract class Advancer {
}
private static class StringFast extends Advancer {
- public static final StringFast instance = new StringFast();
- private static final Schema s = Schema.create(Schema.Type.STRING);
+ public static final StringFast INSTANCE = new StringFast();
+ private static final Schema S = Schema.create(Schema.Type.STRING);
private StringFast() {
- super(s, s);
+ super(S, S);
}
public String nextString(Decoder in) throws IOException {
@@ -499,11 +499,11 @@ abstract class Advancer {
}
private static class BytesFast extends Advancer {
- public static final BytesFast instance = new BytesFast();
- private static final Schema s = Schema.create(Schema.Type.BYTES);
+ public static final BytesFast INSTANCE = new BytesFast();
+ private static final Schema S = Schema.create(Schema.Type.BYTES);
private BytesFast() {
- super(s, s);
+ super(S, S);
}
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
@@ -553,7 +553,7 @@ abstract class Advancer {
//// to the underlying value read.
private static class LongFromInt extends Advancer {
- public static final LongFromInt instance = new LongFromInt();
+ public static final LongFromInt INSTANCE = new LongFromInt();
private LongFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.LONG));
@@ -569,7 +569,7 @@ abstract class Advancer {
}
private static class FloatFromInt extends Advancer {
- public static final FloatFromInt instance = new FloatFromInt();
+ public static final FloatFromInt INSTANCE = new FloatFromInt();
private FloatFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.FLOAT));
@@ -585,7 +585,7 @@ abstract class Advancer {
}
private static class FloatFromLong extends Advancer {
- public static final FloatFromLong instance = new FloatFromLong();
+ public static final FloatFromLong INSTANCE = new FloatFromLong();
private FloatFromLong() {
super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.FLOAT));
@@ -601,7 +601,7 @@ abstract class Advancer {
}
private static class DoubleFromInt extends Advancer {
- public static final DoubleFromInt instance = new DoubleFromInt();
+ public static final DoubleFromInt INSTANCE = new DoubleFromInt();
private DoubleFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.DOUBLE));
@@ -617,7 +617,7 @@ abstract class Advancer {
}
private static class DoubleFromLong extends Advancer {
- public static final DoubleFromLong instance = new DoubleFromLong();
+ public static final DoubleFromLong INSTANCE = new DoubleFromLong();
private DoubleFromLong() {
super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.DOUBLE));
@@ -633,7 +633,7 @@ abstract class Advancer {
}
private static class DoubleFromFloat extends Advancer {
- public static final DoubleFromFloat instance = new DoubleFromFloat();
+ public static final DoubleFromFloat INSTANCE = new DoubleFromFloat();
private DoubleFromFloat() {
super(Schema.create(Schema.Type.FLOAT), Schema.create(Schema.Type.DOUBLE));
@@ -649,7 +649,7 @@ abstract class Advancer {
}
private static class BytesFromString extends Advancer {
- public static final BytesFromString instance = new BytesFromString();
+ public static final BytesFromString INSTANCE = new BytesFromString();
private BytesFromString() {
super(Schema.create(Schema.Type.STRING), Schema.create(Schema.Type.BYTES));
@@ -666,7 +666,7 @@ abstract class Advancer {
}
private static class StringFromBytes extends Advancer {
- public static final StringFromBytes instance = new StringFromBytes();
+ public static final StringFromBytes INSTANCE = new StringFromBytes();
private StringFromBytes() {
super(Schema.create(Schema.Type.BYTES), Schema.create(Schema.Type.STRING));
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
index 5ff9e50..19c77a8 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -143,4 +143,5 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
public void setSchema(Schema s) {
throw new UnsupportedOperationException();
}
-}
\ No newline at end of file
+}
+
[avro] 06/14: More tweaks
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit cc71ce4abd087254403706b4c348b9169e92e3b4
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 01:17:40 2019 -0700
More tweaks
---
.../java/org/apache/avro/generic/Advancer.java | 40 ++++++++++------------
1 file changed, 19 insertions(+), 21 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 9064a39..d75e9ee 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -470,9 +470,6 @@ abstract class Advancer {
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
return branches[branch];
}
- public Object next(Decoder in) throws IOException {
- return branches[in.readIndex()].next(in);
- }
}
private static class ReaderUnion extends Advancer {
@@ -485,9 +482,6 @@ abstract class Advancer {
throw new IllegalArgumentException("Branch much be " + branch + ", got " + b);
return advancer;
}
- public Object next(Decoder in) throws IOException {
- return advancer.next(in);
- }
}
@@ -503,27 +497,29 @@ abstract class Advancer {
* reader's fields. Thus, the following is how to read a record:
* <pre>
* for (int i = 0; i < a.advancers.length; i++) {
- * dataum.set(a.readerOrder[i], a.advancers[i].next());
+ * dataum.set(a.readerOrder[i].pos(), a.advancers[i].next(in));
* }
- * a.done();
+ * a.done(in);
* </pre>
* Note that a decoder <em>must</em> call {@link done} after interpreting
* all the elemnts in {@link advancers}.
*
* As a convenience, {@link inOrder} is set to true iff the reader
- * and writer order agrees (i.e., iff <code>readerOrder[i] ==
+ * and writer order agrees (i.e., iff <code>readerOrder[i].pos() ==
* i</code> for all i). Generated code can use this to optimize this
* common case. */
public static class Record extends Advancer {
public final Advancer[] advancers;
private Schema[] finalSkips;
- public final int[] readerOrder;
+ public final Schema.Field[] readerOrder;
public final boolean inOrder;
- private Record(Advancer[] advancers, Schema[] finalSkips, int[] order, boolean inOrder) {
+ private Record(Advancer[] advancers, Schema[] finalSkips,
+ Schema.Field[] readerOrder, boolean inOrder)
+ {
this.advancers = advancers;
this.finalSkips = finalSkips;
- this.readerOrder = order;
+ this.readerOrder = readerOrder;
this.inOrder = inOrder;
}
@@ -540,8 +536,7 @@ abstract class Advancer {
/** Special subclasses of Advance will encapsulate skipping. */
// Compute the "readerOrder" argument to Advancer.Record constructor
- int[] readOrder = new int[ra.readerOrder.length];
- for (int i = 0; i < readOrder.length; i++) readOrder[i] = ra.readerOrder[i].pos();
+ Schema.Field[] readOrder = ra.readerOrder;
// Compute the "advancers" argument to Advancer.Record constructor
Advancer[] fieldAdvs = new Advancer[readOrder.length];
@@ -573,15 +568,21 @@ abstract class Advancer {
// record's field order not changing.
boolean inOrder = true;
for (int k = 0; k < ra.firstDefault-1; k++)
- inOrder &= (readOrder[k] < readOrder[k+1]);
+ inOrder &= (readOrder[k].pos() < readOrder[k+1].pos());
if (inOrder) {
Advancer[] newAdvancers = new Advancer[fieldAdvs.length];
+ Schema.Field[] newReadOrder = new Schema.Field[fieldAdvs.length];
for (int k = 0, rf2 = 0, df = ra.firstDefault; k < readOrder.length; k++) {
- if (readOrder[rf2] < readOrder[df]) newAdvancers[k] = fieldAdvs[rf2++];
- else newAdvancers[k] = fieldAdvs[df++];
+ if (readOrder[rf2].pos() < readOrder[df].pos()) {
+ newAdvancers[k] = fieldAdvs[rf2];
+ newReadOrder[k] = readOrder[rf2++];
+ } else {
+ newAdvancers[k] = fieldAdvs[df];
+ newReadOrder[k] = readOrder[df++];
+ }
}
- for (int k = 0; k < readOrder.length; k++) readOrder[k] = k;
fieldAdvs = newAdvancers;
+ readOrder = newReadOrder;
}
return new Record(fieldAdvs, finalSkips, readOrder, inOrder);
@@ -596,9 +597,6 @@ abstract class Advancer {
this.field = field;
}
- public Object next(Decoder in) throws IOException
- { ignore(toSkip, in); return field.next(in); }
-
public Object nextNull(Decoder in) throws IOException
{ ignore(toSkip, in); return field.nextNull(in); }
[avro] 07/14: Adding alternative to GenericDatumeReader which uses
Advancers.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit e23abae419376f4c69f6f13c2d25bd6afedb7474
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 01:36:29 2019 -0700
Adding alternative to GenericDatumeReader which uses Advancers.
---
.../apache/avro/generic/GenericDatumReader2.java | 79 ++++++++++++++++++++++
1 file changed, 79 insertions(+)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
new file mode 100644
index 0000000..d86b038
--- /dev/null
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -0,0 +1,79 @@
+/*
+ * 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.avro.generic;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.avro.Resolver;
+import org.apache.avro.Schema;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.io.Decoder;
+
+public class GenericDatumReader2<D> implements DatumReader<D> {
+ private final Advancer.Record advancer;
+ private final GenericData data;
+
+ private GenericDatumReader2(Advancer.Record a, GenericData d) {
+ advancer = a;
+ data = d;
+ }
+
+ public static GenericDatumReader2 getReaderFor(Schema writer, Schema reader, GenericData data) {
+ // TODO: add caching
+ Resolver.Action a = Resolver.resolve(writer, reader, data);
+ Advancer.Record r = (Advancer.Record)Advancer.from(a);
+ return new GenericDatumReader2(r, data);
+ }
+
+ public D read(D reuse, Decoder in) throws IOException {
+ for (int i = 0; i < advancer.advancers.length; i++) {
+ int p = advancer.readerOrder[i].pos();
+ reuse.set(p, read(in, null, advancer.advancers[i]));
+ }
+ advancer.done(in);
+ }
+
+ public Object read(Object reuse, Schema expected, Advancer a, Decoder in)
+ throws IOException
+ {
+ switch (expected.getType()) {
+ case NULL: return a.nextNull(in);
+ case BOOLEAN: return (Boolean) a.nextBoolean(in);
+ case INT: return (Integer) a.nextInt(in);
+ case LONG: return (Long) a.nextLong(in);
+ case FLOAT: return (Float) a.nextFloat(in);
+ case DOUBLE: return (Double) a.nextDouble(in);
+ case STRING: return (String) a.nextString(in);
+ case BYTES: return a.nextBytes(in, (ByteBuffer)reuse);
+ case FIXED:
+ case ARRAY:
+ case MAP:
+ case RECORD:
+ case UNION:
+ default:
+ throw new IllegalArgumentException("Can't handle this yet.");
+ }
+ }
+
+ /** Throws {@link UnsupportedOperationException}. (In retrospect, making
+ * DatumReaders mutable wasn't a good idea...) */
+ public void setSchema(Schema s) {
+ throw new UnsupportedOperationException();
+ }
+}
\ No newline at end of file
[avro] 09/14: Advancer and GenericDatumeReader2 now compile
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 3fe8df68990f145380c3acee237c5d7539a21a0f
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 12:24:47 2019 -0700
Advancer and GenericDatumeReader2 now compile
---
.../java/org/apache/avro/generic/Advancer.java | 8 +++----
.../apache/avro/generic/GenericDatumReader2.java | 28 ++++++++++++++++++----
2 files changed, 26 insertions(+), 10 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 4fe3d7b..8f2b010 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -223,7 +223,7 @@ abstract class Advancer {
* to use to read an array of int:
*
* <pre>
- * Advancer.Container c = advancer.getContainerAdvancer(in);
+ * Advancer c = advancer.getContainerAdvancer(in);
* Advancer.Container ec = c.getElementAdvancer(in);
* for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
* for (long j = 0; j < i; j++) {
@@ -244,8 +244,7 @@ abstract class Advancer {
}
private static class ArrayContainer extends Container {
- private final Advancer elementAdvancer;
- public ArrayContainer(Advancer elementAdvancer) { super(elementAdvancer); }
+ public ArrayContainer(Advancer ea) { super(ea); }
public long firstChunk(Decoder in) throws IOException
{ return in.readArrayStart(); }
public long nextChunk(Decoder in) throws IOException
@@ -253,8 +252,7 @@ abstract class Advancer {
}
private static class MapContainer extends Container {
- private final Advancer elementAdvancer;
- public MapContainer(Advancer elementAdvancer) { super(elementAdvancer); }
+ public MapContainer(Advancer ea) { super(ea); }
public long firstChunk(Decoder in) throws IOException
{ return in.readMapStart(); }
public long nextChunk(Decoder in) throws IOException
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
index d86b038..46506c2 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -19,17 +19,21 @@ package org.apache.avro.generic;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.List;
import org.apache.avro.Resolver;
import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
-public class GenericDatumReader2<D> implements DatumReader<D> {
+public class GenericDatumReader2<D extends IndexedRecord> implements DatumReader<D> {
+ private final Schema reader, writer;
private final Advancer.Record advancer;
private final GenericData data;
- private GenericDatumReader2(Advancer.Record a, GenericData d) {
+ private GenericDatumReader2(Schema writer, Schema reader, Advancer.Record a, GenericData d) {
+ this.writer = writer;
+ this.reader = reader;
advancer = a;
data = d;
}
@@ -38,15 +42,18 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
// TODO: add caching
Resolver.Action a = Resolver.resolve(writer, reader, data);
Advancer.Record r = (Advancer.Record)Advancer.from(a);
- return new GenericDatumReader2(r, data);
+ return new GenericDatumReader2(writer, reader, r, data);
}
public D read(D reuse, Decoder in) throws IOException {
+ List<Schema.Field> wf = writer.getFields();
+ if (reuse == null) reuse = null; // FIXME
for (int i = 0; i < advancer.advancers.length; i++) {
int p = advancer.readerOrder[i].pos();
- reuse.set(p, read(in, null, advancer.advancers[i]));
+ reuse.put(p, read(null, wf.get(i).schema(), advancer.advancers[i], in));
}
advancer.done(in);
+ return reuse;
}
public Object read(Object reuse, Schema expected, Advancer a, Decoder in)
@@ -62,7 +69,18 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
case STRING: return (String) a.nextString(in);
case BYTES: return a.nextBytes(in, (ByteBuffer)reuse);
case FIXED:
- case ARRAY:
+ case ARRAY: {
+ List result = null; // FIXME -- use GenericData methods here...
+ Advancer.Container c = advancer.getContainerAdvancer(in);
+ Advancer ec = c.getElementAdvancer(in);
+ Schema es = expected.getElementType();
+ for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ for (long j = 0; j < i; j++) {
+ result.add(read(null, es, ec, in));
+ }
+ }
+ }
+
case MAP:
case RECORD:
case UNION:
[avro] 12/14: formatting changes
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit cefbf862a7ba5f024e990a987357926a3f8ab2be
Author: Thiruvalluvan M G <th...@startsmartlabs.com>
AuthorDate: Tue Apr 30 19:48:50 2019 -0700
formatting changes
---
.../src/main/java/org/apache/avro/Resolver.java | 54 +-
.../java/org/apache/avro/generic/Advancer.java | 975 ++++++++++++++-------
.../apache/avro/generic/GenericDatumReader2.java | 63 +-
3 files changed, 728 insertions(+), 364 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
index 7c9e2c3..a01dfb9 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
@@ -163,11 +163,11 @@ public class Resolver {
}
/**
- * In this case, there's nothing to be done for resolution: the two
- * schemas are effectively the same. This action will be generated
- * <em>only</em> for primitive types (incl. STRING and BYTES) and
- * also for fixed types, and <not>not</not for any other kind of
- * schema. */
+ * In this case, there's nothing to be done for resolution: the two schemas are
+ * effectively the same. This action will be generated <em>only</em> for
+ * primitive types (incl. STRING and BYTES) and also for fixed types, and
+ * <not>not</not for any other kind of schema.
+ */
public static class DoNothing extends Action {
public DoNothing(Schema w, Schema r, GenericData d) {
super(w, r, d, Action.Type.DO_NOTHING);
@@ -251,9 +251,7 @@ public class Resolver {
*/
public static class Promote extends Action {
public static enum Promotions {
- INT2LONG, INT2FLOAT, INT2DOUBLE,
- LONG2FLOAT, LONG2DOUBLE, FLOAT2DOUBLE,
- STRING2BYTES, BYTES2STRING
+ INT2LONG, INT2FLOAT, INT2DOUBLE, LONG2FLOAT, LONG2DOUBLE, FLOAT2DOUBLE, STRING2BYTES, BYTES2STRING
}
/** The exact promotion being represented. */
@@ -283,9 +281,8 @@ public class Resolver {
}
/**
- * Returns the {@link Promtion} that would resolve schema
- * <code>w</code> against <code>r</code>, or null if there is no
- * such promotion.
+ * Returns the {@link Promtion} that would resolve schema <code>w</code> against
+ * <code>r</code>, or null if there is no such promotion.
*/
private static Promotions findPromotion(Schema w, Schema r) {
if (w.getType() == r.getType())
@@ -294,27 +291,35 @@ public class Resolver {
switch (r.getType()) {
case LONG:
switch (wt) {
- case INT: return Promotions.INT2LONG;
+ case INT:
+ return Promotions.INT2LONG;
}
break;
case FLOAT:
switch (wt) {
- case INT: return Promotions.INT2FLOAT;
- case LONG: return Promotions.LONG2FLOAT;
+ case INT:
+ return Promotions.INT2FLOAT;
+ case LONG:
+ return Promotions.LONG2FLOAT;
}
break;
case DOUBLE:
switch (wt) {
- case INT: return Promotions.INT2DOUBLE;
- case LONG: return Promotions.LONG2DOUBLE;
- case FLOAT: return Promotions.FLOAT2DOUBLE;
+ case INT:
+ return Promotions.INT2DOUBLE;
+ case LONG:
+ return Promotions.LONG2DOUBLE;
+ case FLOAT:
+ return Promotions.FLOAT2DOUBLE;
}
break;
case BYTES:
- if (wt == Schema.Type.STRING) return Promotions.STRING2BYTES;
+ if (wt == Schema.Type.STRING)
+ return Promotions.STRING2BYTES;
break;
case STRING:
- if (wt == Schema.Type.STRING) return Promotions.BYTES2STRING;
+ if (wt == Schema.Type.STRING)
+ return Promotions.BYTES2STRING;
break;
}
return null;
@@ -422,12 +427,11 @@ public class Resolver {
public final Action[] fieldActions;
/**
- * Contains (all of) the reader's fields. The first {@link
- * firstDefault} of these are the fields that will be read from
- * the writer: these are in the order dictated by writer's
- * schema. The remaining fields from {@link firstDefault} to the
- * end of the array will be read from default values (actions for
- * these default values are found in {@link defaults}. Note that
+ * Contains (all of) the reader's fields. The first {@link firstDefault} of
+ * these are the fields that will be read from the writer: these are in the
+ * order dictated by writer's schema. The remaining fields from
+ * {@link firstDefault} to the end of the array will be read from default values
+ * (actions for these default values are found in {@link defaults}. Note that
* the default fields are in the order of the reader's schema.
*/
public final Field[] readerOrder;
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index 7c469cb..21bb22a 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -27,67 +27,108 @@ import org.apache.avro.Schema;
import org.apache.avro.io.Decoder;
import org.apache.avro.util.Utf8;
-
-/** An "Advancer" is a tree of objects that apply resolution logic
- * while reading values out of a {@link Decoder}.
- *
- * An Advancer tree is created by calling {@link advancerFor} on a
- * {@link Resolver.Action} object. The resulting tree mimics the
- * reader schema of that Action object.
- *
- * A decoder for that reader schema is meant to traverse the schema
- * in a depth-first fashion. When it hits a leaf of type
- * <code>Xyz</code>, it should call corresponding
- * <code>nextXyx</code> on the Advancer. For example, if the reader
- * hits a lead indicating that an integer should be read, it should
- * call {@link nextInt}, as in <code>a.nextInt(in)</code>, where
- * <code>a</code> is the advancer being traversed, and
- * <code>in</code> is the Decoder being read from.
- *
- * When traversing an Array or Map in the reader schema, the decoder
- * should call {@link getElementAdvancer} to retrieve the advancer
- * object for the contained element-schema or value-schema. See the
- * JavaDoc for {@link getElementAdvancer} for instructions on how to
- * decode these types.
- *
- * For unions, the decoder should call {@link nextIndex} to fetch the
- * branch and then {@link getBranchAdvancer} to get the advancer of
- * that branch. (Calling {@link next} on a union will read the
- * index, pick the right advancer based on the index, and then read
- * and return the actual value.)
- *
- * Traversing records, arrays, and maps is more involved. In the
- * case of an array or map, call {@link getArrayAdvancer} {@link
- * getMapAdvancer} and proceed as described in the documentation for
- * {@link Advancer.Container}. For records, best to just look at the
- * implementation of {@link GenericDatumReader2}.
- **/
+/**
+ * An "Advancer" is a tree of objects that apply resolution logic while reading
+ * values out of a {@link Decoder}.
+ *
+ * An Advancer tree is created by calling {@link advancerFor} on a
+ * {@link Resolver.Action} object. The resulting tree mimics the reader schema
+ * of that Action object.
+ *
+ * A decoder for that reader schema is meant to traverse the schema in a
+ * depth-first fashion. When it hits a leaf of type <code>Xyz</code>, it should
+ * call corresponding <code>nextXyx</code> on the Advancer. For example, if the
+ * reader hits a lead indicating that an integer should be read, it should call
+ * {@link nextInt}, as in <code>a.nextInt(in)</code>, where <code>a</code> is
+ * the advancer being traversed, and <code>in</code> is the Decoder being read
+ * from.
+ *
+ * When traversing an Array or Map in the reader schema, the decoder should call
+ * {@link getElementAdvancer} to retrieve the advancer object for the contained
+ * element-schema or value-schema. See the JavaDoc for
+ * {@link getElementAdvancer} for instructions on how to decode these types.
+ *
+ * For unions, the decoder should call {@link nextIndex} to fetch the branch and
+ * then {@link getBranchAdvancer} to get the advancer of that branch. (Calling
+ * {@link next} on a union will read the index, pick the right advancer based on
+ * the index, and then read and return the actual value.)
+ *
+ * Traversing records, arrays, and maps is more involved. In the case of an
+ * array or map, call {@link getArrayAdvancer} {@link getMapAdvancer} and
+ * proceed as described in the documentation for {@link Advancer.Container}. For
+ * records, best to just look at the implementation of
+ * {@link GenericDatumReader2}.
+ **/
abstract class Advancer {
protected Exception exception() {
throw new UnsupportedOperationException();
}
- //// API methods of Advancer. Used by decoding methods to
+ //// API methods of Advancer. Used by decoding methods to
//// read values out of Decoder, applying resolution logic
- //// in the process. In the base class, these do throw
- //// a not-supported exception. Specific subclasses implement
+ //// in the process. In the base class, these do throw
+ //// a not-supported exception. Specific subclasses implement
//// certain ones, e.g., IntFast (the Advancer used when
//// an integer is read with no promotion) overrides just
//// readInt.
public final Schema writer, reader;
- protected Advancer(Schema w, Schema r) { this.writer = w; this.reader = r; }
-
- public Object next(Decoder in) throws IOException { exception(); return null; }
- public Object nextNull(Decoder in) throws IOException { exception(); return null; }
- public boolean nextBoolean(Decoder in) throws IOException { exception(); return false; }
- public int nextInt(Decoder in) throws IOException { exception(); return 0; }
- public long nextLong(Decoder in) throws IOException { exception(); return 0; }
- public float nextFloat(Decoder in) throws IOException { exception(); return 0; }
- public double nextDouble(Decoder in) throws IOException { exception(); return 0; }
- public int nextEnum(Decoder in) throws IOException { exception(); return 0; }
- public Utf8 nextString(Decoder in, Utf8 old) throws IOException { exception(); return null; }
- public String nextString(Decoder in) throws IOException { exception(); return null; }
+
+ protected Advancer(Schema w, Schema r) {
+ this.writer = w;
+ this.reader = r;
+ }
+
+ public Object next(Decoder in) throws IOException {
+ exception();
+ return null;
+ }
+
+ public Object nextNull(Decoder in) throws IOException {
+ exception();
+ return null;
+ }
+
+ public boolean nextBoolean(Decoder in) throws IOException {
+ exception();
+ return false;
+ }
+
+ public int nextInt(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
+
+ public long nextLong(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
+
+ public float nextFloat(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
+
+ public double nextDouble(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
+
+ public int nextEnum(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
+
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ exception();
+ return null;
+ }
+
+ public String nextString(Decoder in) throws IOException {
+ exception();
+ return null;
+ }
+
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
exception();
return null;
@@ -103,10 +144,15 @@ abstract class Advancer {
}
/** Get index for a union. */
- public int nextIndex(Decoder in) throws IOException { exception(); return 0; }
+ public int nextIndex(Decoder in) throws IOException {
+ exception();
+ return 0;
+ }
- /** Access to contained advancer for unions. You must call {@link
- * nextIndex} before calling this method. */
+ /**
+ * Access to contained advancer for unions. You must call {@link nextIndex}
+ * before calling this method.
+ */
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
exception();
return null;
@@ -130,72 +176,92 @@ abstract class Advancer {
return null;
}
-
- ////// Here's the builder for Advancer trees. The subclasses used by
+ ////// Here's the builder for Advancer trees. The subclasses used by
////// this implementation are found below.
- /** Build a {@link Advancer} tree that for a given {@link
- * Resolver.Action} tree. If input argument (<code>a</code>) is a
- * {@link Resolver.RecordAdjust}, the result is guaranteed to be a
- * {@link Advancer.Record}. */
+ /**
+ * Build a {@link Advancer} tree that for a given {@link Resolver.Action} tree.
+ * If input argument (<code>a</code>) is a {@link Resolver.RecordAdjust}, the
+ * result is guaranteed to be a {@link Advancer.Record}.
+ */
public static Advancer from(Resolver.Action a) {
switch (a.type) {
case DO_NOTHING:
switch (a.reader.getType()) {
- case NULL: return NullFast.instance;
- case BOOLEAN: return BooleanFast.instance;
- case INT: return IntFast.instance;
- case LONG: return LongFast.instance;
- case FLOAT: return FloatFast.instance;
- case DOUBLE: return DoubleFast.instance;
- case STRING: return StringFast.instance;
- case BYTES: return BytesFast.instance;
- case FIXED: return new FixedFast(a.writer, a.reader);
+ case NULL:
+ return NullFast.instance;
+ case BOOLEAN:
+ return BooleanFast.instance;
+ case INT:
+ return IntFast.instance;
+ case LONG:
+ return LongFast.instance;
+ case FLOAT:
+ return FloatFast.instance;
+ case DOUBLE:
+ return DoubleFast.instance;
+ case STRING:
+ return StringFast.instance;
+ case BYTES:
+ return BytesFast.instance;
+ case FIXED:
+ return new FixedFast(a.writer, a.reader);
default:
throw new IllegalArgumentException("Unexpected schema for DoNothing:" + a.reader);
}
case PROMOTE:
- switch (((Resolver.Promote)a).promotion) {
- case INT2LONG: return LongFromInt.instance;
- case INT2FLOAT: return FloatFromInt.instance;
- case INT2DOUBLE: return DoubleFromInt.instance;
- case LONG2FLOAT: return FloatFromLong.instance;
- case LONG2DOUBLE: return DoubleFromLong.instance;
- case FLOAT2DOUBLE: return DoubleFromFloat.instance;
- case STRING2BYTES: return BytesFromString.instance;
- case BYTES2STRING: return StringFromBytes.instance;
+ switch (((Resolver.Promote) a).promotion) {
+ case INT2LONG:
+ return LongFromInt.instance;
+ case INT2FLOAT:
+ return FloatFromInt.instance;
+ case INT2DOUBLE:
+ return DoubleFromInt.instance;
+ case LONG2FLOAT:
+ return FloatFromLong.instance;
+ case LONG2DOUBLE:
+ return DoubleFromLong.instance;
+ case FLOAT2DOUBLE:
+ return DoubleFromFloat.instance;
+ case STRING2BYTES:
+ return BytesFromString.instance;
+ case BYTES2STRING:
+ return StringFromBytes.instance;
default:
throw new IllegalArgumentException("Unexpected promotion:" + a);
}
case ENUM:
- Resolver.EnumAdjust e = (Resolver.EnumAdjust)a;
- if (e.noAdjustmentsNeeded) return new EnumFast(a.writer, a.reader);
- else return new EnumWithAdjustments(a.writer, a.reader, e.adjustments);
+ Resolver.EnumAdjust e = (Resolver.EnumAdjust) a;
+ if (e.noAdjustmentsNeeded)
+ return new EnumFast(a.writer, a.reader);
+ else
+ return new EnumWithAdjustments(a.writer, a.reader, e.adjustments);
case CONTAINER:
- Advancer ea = Advancer.from(((Resolver.Container)a).elementAction);
- if (a.writer.getType() == Schema.Type.ARRAY)
- return new Container(a.writer, a.reader, ea);
- else return new Map(a.writer, a.reader, ea);
+ Advancer ea = Advancer.from(((Resolver.Container) a).elementAction);
+ if (a.writer.getType() == Schema.Type.ARRAY)
+ return new Container(a.writer, a.reader, ea);
+ else
+ return new Map(a.writer, a.reader, ea);
case RECORD:
- return Advancer.Record.from((Resolver.RecordAdjust)a);
+ return Advancer.Record.from((Resolver.RecordAdjust) a);
case WRITER_UNION:
- Resolver.WriterUnion wu = (Resolver.WriterUnion)a;
+ Resolver.WriterUnion wu = (Resolver.WriterUnion) a;
Advancer[] branches = new Advancer[wu.actions.length];
for (int i = 0; i < branches.length; i++)
branches[i] = Advancer.from(wu.actions[i]);
- if (wu.unionEquiv) return new EquivUnion(a.writer, a.reader, branches);
+ if (wu.unionEquiv)
+ return new EquivUnion(a.writer, a.reader, branches);
return new WriterUnion(a.writer, a.reader, branches);
case READER_UNION:
- Resolver.ReaderUnion ru = (Resolver.ReaderUnion)a;
- return new ReaderUnion(a.writer, a.reader,
- ru.firstMatch, Advancer.from(ru.actualAction));
+ Resolver.ReaderUnion ru = (Resolver.ReaderUnion) a;
+ return new ReaderUnion(a.writer, a.reader, ru.firstMatch, Advancer.from(ru.actualAction));
case ERROR:
- return new Error(a.writer,a.reader, a.toString());
+ return new Error(a.writer, a.reader, a.toString());
case SKIP:
throw new RuntimeException("Internal error. Skip should've been consumed.");
default:
@@ -215,167 +281,253 @@ abstract class Advancer {
}
return result;
}
+
private static final Schema[] EMPTY_SCHEMA_ARRAY = new Schema[0];
////// Subclasses of Advancer -- real work is done in these
- /** All methods of <code>this</code> throw {@link
- * AvroTypeException} with appropriate message. Used for
- * throwing resolution errors in a lazy fashion (i.e., as actual
- * data causes the error to manifest). */
+ /**
+ * All methods of <code>this</code> throw {@link AvroTypeException} with
+ * appropriate message. Used for throwing resolution errors in a lazy fashion
+ * (i.e., as actual data causes the error to manifest).
+ */
private static class Error extends Advancer {
String msg;
- public Error(Schema w, Schema r, String msg) { super(w,r); this.msg = msg; }
+
+ public Error(Schema w, Schema r, String msg) {
+ super(w, r);
+ this.msg = msg;
+ }
+
protected Exception exception() {
throw new AvroTypeException(msg);
}
}
- /** Used for Array. The following fragment illustrates how
- * to use to read an array of int:
- *
- * <pre>
- * Advancer.Container c = advancer.getArrayAdvancer(in);
- * for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
- * for (long j = 0; j < i; j++) {
- * int element = c.elementAdvancer.readInt(in);
- * // .. do something with this element
- * }
- * }
- * </pre>
- * See the implementation of {@link GenericDatumReader2} for more
- * illustrations. */
+ /**
+ * Used for Array. The following fragment illustrates how to use to read an
+ * array of int:
+ *
+ * <pre>
+ * Advancer.Container c = advancer.getArrayAdvancer(in);
+ * for (long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ * for (long j = 0; j < i; j++) {
+ * int element = c.elementAdvancer.readInt(in);
+ * // .. do something with this element
+ * }
+ * }
+ * </pre>
+ *
+ * See the implementation of {@link GenericDatumReader2} for more illustrations.
+ */
public static class Container extends Advancer {
public final Advancer elementAdvancer;
- public Container(Schema w, Schema r, Advancer ea)
- { super(w,r); elementAdvancer = ea; }
-
- public long firstChunk(Decoder in) throws IOException
- { return in.readArrayStart(); }
-
- public long nextChunk(Decoder in) throws IOException
- { return in.arrayNext(); }
- }
-
- /** Used for Map. The following fragment illustrates how
- * to use to read an array of int:
- *
- * <pre>
- * Advancer.Map c = advancer.getMapAdvancer(in);
- * for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
- * for (long j = 0; j < i; j++) {
- * String key = c.keyAdvancer.readString(in);
- * int element = c.elementAdvancer.readInt(in);
- * // .. do something with this element
- * }
- * }
- * </pre>
- * See the implementation of {@link GenericDatumReader2} for more
- * illustrations. */
+
+ public Container(Schema w, Schema r, Advancer ea) {
+ super(w, r);
+ elementAdvancer = ea;
+ }
+
+ public long firstChunk(Decoder in) throws IOException {
+ return in.readArrayStart();
+ }
+
+ public long nextChunk(Decoder in) throws IOException {
+ return in.arrayNext();
+ }
+ }
+
+ /**
+ * Used for Map. The following fragment illustrates how to use to read an array
+ * of int:
+ *
+ * <pre>
+ * Advancer.Map c = advancer.getMapAdvancer(in);
+ * for (long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ * for (long j = 0; j < i; j++) {
+ * String key = c.keyAdvancer.readString(in);
+ * int element = c.elementAdvancer.readInt(in);
+ * // .. do something with this element
+ * }
+ * }
+ * </pre>
+ *
+ * See the implementation of {@link GenericDatumReader2} for more illustrations.
+ */
public static class Map extends Container {
public final Advancer keyAdvancer = StringFast.instance;
- public Map(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
- public long firstChunk(Decoder in) throws IOException
- { return in.readMapStart(); }
+ public Map(Schema w, Schema r, Advancer ea) {
+ super(w, r, ea);
+ }
- public long nextChunk(Decoder in) throws IOException
- { return in.mapNext(); }
+ public long firstChunk(Decoder in) throws IOException {
+ return in.readMapStart();
+ }
+
+ public long nextChunk(Decoder in) throws IOException {
+ return in.mapNext();
+ }
}
//// The following set of subclasses are for when there is no
- //// resolution logic to be applied. All that needs to be done
+ //// resolution logic to be applied. All that needs to be done
//// is call the corresponding method on the Decoder.
private static class NullFast extends Advancer {
public static final NullFast instance = new NullFast();
private static final Schema s = Schema.create(Schema.Type.NULL);
- private NullFast() { super(s,s); }
+
+ private NullFast() {
+ super(s, s);
+ }
+
public Object nextNull(Decoder in) throws IOException {
- in.readNull();
+ in.readNull();
return null;
}
- public Object next(Decoder in) throws IOException { return nextNull(in); }
+
+ public Object next(Decoder in) throws IOException {
+ return nextNull(in);
+ }
}
private static class BooleanFast extends Advancer {
public static final BooleanFast instance = new BooleanFast();
private static final Schema s = Schema.create(Schema.Type.BOOLEAN);
- private BooleanFast() { super(s,s); }
+
+ private BooleanFast() {
+ super(s, s);
+ }
+
public boolean nextBoolean(Decoder in) throws IOException {
- return in.readBoolean();
+ return in.readBoolean();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextBoolean(in);
}
- public Object next(Decoder in) throws IOException { return nextBoolean(in); }
}
private static class IntFast extends Advancer {
public static final IntFast instance = new IntFast();
private static final Schema s = Schema.create(Schema.Type.INT);
- private IntFast() { super(s,s); }
+
+ private IntFast() {
+ super(s, s);
+ }
+
public int nextInt(Decoder in) throws IOException {
- return in.readInt();
+ return in.readInt();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextInt(in);
}
- public Object next(Decoder in) throws IOException { return nextInt(in); }
}
private static class LongFast extends Advancer {
public static final LongFast instance = new LongFast();
private static final Schema s = Schema.create(Schema.Type.LONG);
- private LongFast() { super(s,s); }
+
+ private LongFast() {
+ super(s, s);
+ }
+
public long nextLong(Decoder in) throws IOException {
- return in.readLong();
+ return in.readLong();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextLong(in);
}
- public Object next(Decoder in) throws IOException { return nextLong(in); }
}
private static class FloatFast extends Advancer {
public static final FloatFast instance = new FloatFast();
private static final Schema s = Schema.create(Schema.Type.FLOAT);
- private FloatFast() { super(s,s); }
+
+ private FloatFast() {
+ super(s, s);
+ }
+
public float nextFloat(Decoder in) throws IOException {
- return in.readFloat();
+ return in.readFloat();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextFloat(in);
}
- public Object next(Decoder in) throws IOException { return nextFloat(in); }
}
private static class DoubleFast extends Advancer {
public static final DoubleFast instance = new DoubleFast();
private static final Schema s = Schema.create(Schema.Type.DOUBLE);
- private DoubleFast() { super(s,s); }
+
+ private DoubleFast() {
+ super(s, s);
+ }
+
public double nextDouble(Decoder in) throws IOException {
- return in.readDouble();
+ return in.readDouble();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextDouble(in);
}
- public Object next(Decoder in) throws IOException { return nextDouble(in); }
}
private static class StringFast extends Advancer {
public static final StringFast instance = new StringFast();
private static final Schema s = Schema.create(Schema.Type.STRING);
- private StringFast() { super(s,s); }
- public String nextString(Decoder in) throws IOException { return in.readString(); }
+
+ private StringFast() {
+ super(s, s);
+ }
+
+ public String nextString(Decoder in) throws IOException {
+ return in.readString();
+ }
+
public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
return in.readString(old);
}
- public Object next(Decoder in) throws IOException { return nextString(in); }
+
+ public Object next(Decoder in) throws IOException {
+ return nextString(in);
+ }
}
private static class BytesFast extends Advancer {
public static final BytesFast instance = new BytesFast();
private static final Schema s = Schema.create(Schema.Type.BYTES);
- private BytesFast() { super(s,s); }
+
+ private BytesFast() {
+ super(s, s);
+ }
+
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
return in.readBytes(old);
}
- public Object next(Decoder in) throws IOException { return nextBytes(in, null); }
+
+ public Object next(Decoder in) throws IOException {
+ return nextBytes(in, null);
+ }
}
private static class FixedFast extends Advancer {
private final int len;
- private FixedFast(Schema w, Schema r) { super(w,r); this.len = w.getFixedSize(); }
+
+ private FixedFast(Schema w, Schema r) {
+ super(w, r);
+ this.len = w.getFixedSize();
+ }
+
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
in.readFixed(bytes, start, len);
return bytes;
}
+
public Object next(Decoder in) throws IOException {
byte[] result = new byte[len];
nextFixed(in, new byte[len]);
@@ -384,9 +536,17 @@ abstract class Advancer {
}
private static class EnumFast extends Advancer {
- public EnumFast(Schema w, Schema r) { super(w,r); }
- public int nextEnum(Decoder in) throws IOException { return in.readEnum(); }
- public Object next(Decoder in) throws IOException { return nextEnum(in); }
+ public EnumFast(Schema w, Schema r) {
+ super(w, r);
+ }
+
+ public int nextEnum(Decoder in) throws IOException {
+ return in.readEnum();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextEnum(in);
+ }
}
//// The following set of subclasses apply promotion logic
@@ -394,160 +554,258 @@ abstract class Advancer {
private static class LongFromInt extends Advancer {
public static final LongFromInt instance = new LongFromInt();
+
private LongFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.LONG));
}
+
public long nextLong(Decoder in) throws IOException {
- return (long) in.readInt();
+ return (long) in.readInt();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextLong(in);
}
- public Object next(Decoder in) throws IOException { return nextLong(in); }
}
private static class FloatFromInt extends Advancer {
public static final FloatFromInt instance = new FloatFromInt();
+
private FloatFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.FLOAT));
}
+
public float nextFloat(Decoder in) throws IOException {
- return (float) in.readInt();
+ return (float) in.readInt();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextFloat(in);
}
- public Object next(Decoder in) throws IOException { return nextFloat(in); }
}
private static class FloatFromLong extends Advancer {
public static final FloatFromLong instance = new FloatFromLong();
+
private FloatFromLong() {
super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.FLOAT));
}
+
public float nextFloat(Decoder in) throws IOException {
- return (long) in.readLong();
+ return (long) in.readLong();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextFloat(in);
}
- public Object next(Decoder in) throws IOException { return nextFloat(in); }
}
private static class DoubleFromInt extends Advancer {
public static final DoubleFromInt instance = new DoubleFromInt();
+
private DoubleFromInt() {
super(Schema.create(Schema.Type.INT), Schema.create(Schema.Type.DOUBLE));
}
+
public double nextDouble(Decoder in) throws IOException {
- return (double) in.readInt();
+ return (double) in.readInt();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextDouble(in);
}
- public Object next(Decoder in) throws IOException { return nextDouble(in); }
}
private static class DoubleFromLong extends Advancer {
public static final DoubleFromLong instance = new DoubleFromLong();
+
private DoubleFromLong() {
super(Schema.create(Schema.Type.LONG), Schema.create(Schema.Type.DOUBLE));
}
+
public double nextDouble(Decoder in) throws IOException {
- return (double) in.readLong();
+ return (double) in.readLong();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextDouble(in);
}
- public Object next(Decoder in) throws IOException { return nextDouble(in); }
}
private static class DoubleFromFloat extends Advancer {
public static final DoubleFromFloat instance = new DoubleFromFloat();
+
private DoubleFromFloat() {
super(Schema.create(Schema.Type.FLOAT), Schema.create(Schema.Type.DOUBLE));
}
+
public double nextDouble(Decoder in) throws IOException {
- return (double) in.readFloat();
+ return (double) in.readFloat();
+ }
+
+ public Object next(Decoder in) throws IOException {
+ return nextDouble(in);
}
- public Object next(Decoder in) throws IOException { return nextDouble(in); }
}
private static class BytesFromString extends Advancer {
public static final BytesFromString instance = new BytesFromString();
+
private BytesFromString() {
super(Schema.create(Schema.Type.STRING), Schema.create(Schema.Type.BYTES));
}
+
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
Utf8 s = in.readString(null);
return ByteBuffer.wrap(s.getBytes(), 0, s.getByteLength());
}
- public Object next(Decoder in) throws IOException { return nextBytes(in, null); }
+
+ public Object next(Decoder in) throws IOException {
+ return nextBytes(in, null);
+ }
}
private static class StringFromBytes extends Advancer {
public static final StringFromBytes instance = new StringFromBytes();
+
private StringFromBytes() {
super(Schema.create(Schema.Type.BYTES), Schema.create(Schema.Type.STRING));
}
+
public String nextString(Decoder in) throws IOException {
return new String(in.readBytes(null).array(), StandardCharsets.UTF_8);
}
+
public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
return new Utf8(in.readBytes(null).array());
}
- public Object next(Decoder in) throws IOException { return nextString(in); }
- }
+ public Object next(Decoder in) throws IOException {
+ return nextString(in);
+ }
+ }
//// This last set of advancers are used when more sophisticated
//// adjustmentds are needed
private static class EnumWithAdjustments extends Advancer {
private final int[] adjustments;
+
public EnumWithAdjustments(Schema w, Schema r, int[] adjustments) {
- super(w,r);
+ super(w, r);
this.adjustments = adjustments;
}
+
public int nextEnum(Decoder in) throws IOException {
return adjustments[in.readInt()];
}
- public Object next(Decoder in) throws IOException { return nextEnum(in); }
+
+ public Object next(Decoder in) throws IOException {
+ return nextEnum(in);
+ }
}
- /** In this case, the writer has a union by the reader doesn't, so we
- * consume the tag ourself and call the corresponding advancer. */
+ /**
+ * In this case, the writer has a union by the reader doesn't, so we consume the
+ * tag ourself and call the corresponding advancer.
+ */
private static class WriterUnion extends Advancer {
private Advancer[] branches;
- public WriterUnion(Schema w, Schema r, Advancer[] b) { super(w,r); branches = b; }
- private final Advancer b(Decoder in) throws IOException
- { return branches[in.readIndex()]; }
+ public WriterUnion(Schema w, Schema r, Advancer[] b) {
+ super(w, r);
+ branches = b;
+ }
- public Object next(Decoder in) throws IOException { return b(in).next(in); }
- public Object nextNull(Decoder in) throws IOException { return b(in).nextNull(in); }
- public boolean nextBoolean(Decoder in) throws IOException { return b(in).nextBoolean(in); }
- public int nextInt(Decoder in) throws IOException { return b(in).nextInt(in); }
- public long nextLong(Decoder in) throws IOException { return b(in).nextLong(in); }
- public float nextFloat(Decoder in) throws IOException { return b(in).nextFloat(in); }
- public double nextDouble(Decoder in) throws IOException { return b(in).nextDouble(in); }
- public int nextEnum(Decoder in) throws IOException { return b(in).nextEnum(in); }
- public String nextString(Decoder in) throws IOException { return b(in).nextString(in); }
- public Utf8 nextString(Decoder in, Utf8 old) throws IOException
- { return b(in).nextString(in, old); }
+ private final Advancer b(Decoder in) throws IOException {
+ return branches[in.readIndex()];
+ }
- public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException
- { return b(in).nextBytes(in, old); }
+ public Object next(Decoder in) throws IOException {
+ return b(in).next(in);
+ }
- public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException
- { return b(in).nextFixed(in, bytes, start, length); }
+ public Object nextNull(Decoder in) throws IOException {
+ return b(in).nextNull(in);
+ }
- public int nextIndex(Decoder in) throws IOException { return b(in).nextIndex(in); }
- public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
- { return b(in).getBranchAdvancer(in, branch); }
+ public boolean nextBoolean(Decoder in) throws IOException {
+ return b(in).nextBoolean(in);
+ }
+
+ public int nextInt(Decoder in) throws IOException {
+ return b(in).nextInt(in);
+ }
+
+ public long nextLong(Decoder in) throws IOException {
+ return b(in).nextLong(in);
+ }
+
+ public float nextFloat(Decoder in) throws IOException {
+ return b(in).nextFloat(in);
+ }
- public Container getArrayAdvancer(Decoder in) throws IOException
- { return b(in).getArrayAdvancer(in); }
+ public double nextDouble(Decoder in) throws IOException {
+ return b(in).nextDouble(in);
+ }
- public Map getMapAdvancer(Decoder in) throws IOException
- { return b(in).getMapAdvancer(in); }
+ public int nextEnum(Decoder in) throws IOException {
+ return b(in).nextEnum(in);
+ }
- public Record getRecordAdvancer(Decoder in) throws IOException
- { return b(in).getRecordAdvancer(in); }
+ public String nextString(Decoder in) throws IOException {
+ return b(in).nextString(in);
+ }
+
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ return b(in).nextString(in, old);
+ }
+
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ return b(in).nextBytes(in, old);
+ }
+
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException {
+ return b(in).nextFixed(in, bytes, start, length);
+ }
+
+ public int nextIndex(Decoder in) throws IOException {
+ return b(in).nextIndex(in);
+ }
+
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
+ return b(in).getBranchAdvancer(in, branch);
+ }
+
+ public Container getArrayAdvancer(Decoder in) throws IOException {
+ return b(in).getArrayAdvancer(in);
+ }
+
+ public Map getMapAdvancer(Decoder in) throws IOException {
+ return b(in).getMapAdvancer(in);
+ }
+
+ public Record getRecordAdvancer(Decoder in) throws IOException {
+ return b(in).getRecordAdvancer(in);
+ }
}
- /** In this case, reader and writer have the same union, so let the decoder
- * consume it as a regular union. */
+ /**
+ * In this case, reader and writer have the same union, so let the decoder
+ * consume it as a regular union.
+ */
private static class EquivUnion extends Advancer {
private final Advancer[] branches;
- public EquivUnion(Schema w, Schema r, Advancer[] b) {super(w,r); branches = b; }
- public int nextIndex(Decoder in) throws IOException { return in.readIndex(); }
+ public EquivUnion(Schema w, Schema r, Advancer[] b) {
+ super(w, r);
+ branches = b;
+ }
+
+ public int nextIndex(Decoder in) throws IOException {
+ return in.readIndex();
+ }
+
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
return branches[branch];
}
@@ -556,59 +814,67 @@ abstract class Advancer {
private static class ReaderUnion extends Advancer {
private int branch;
private Advancer advancer;
- public ReaderUnion(Schema w, Schema r, int b, Advancer a)
- { super(w,r); branch = b; advancer = a; }
- public int nextIndex(Decoder in) { return branch; }
+ public ReaderUnion(Schema w, Schema r, int b, Advancer a) {
+ super(w, r);
+ branch = b;
+ advancer = a;
+ }
+
+ public int nextIndex(Decoder in) {
+ return branch;
+ }
public Advancer getBranchAdvancer(Decoder in, int b) {
if (b != this.branch)
- throw new IllegalArgumentException("Branch much be " + branch + ", got " + b);
+ throw new IllegalArgumentException("Branch much be " + branch + ", got " + b);
return advancer;
}
}
-
-
-
//// Records are particularly intricate because we may have to skip
//// fields, read fields out of order, and use default values.
- /** Advancer for records. The {@link advancers} array contains an
- * advancer for each field, ordered according writer (which
- * determines the order in which data must be read). The {@link
- * readerOrder} array tells you how those advancers line up with the
- * reader's fields. Thus, the following is how to read a record:
- * <pre>
- * for (int i = 0; i < a.advancers.length; i++) {
- * dataum.set(a.readerOrder[i].pos(), a.advancers[i].next(in));
- * }
- * a.done(in);
- * </pre>
- * Note that a decoder <em>must</em> call {@link done} after interpreting
- * all the elemnts in {@link advancers}.
- *
- * As a convenience, {@link inOrder} is set to true iff the reader
- * and writer order agrees (i.e., iff <code>readerOrder[i].pos() ==
- * i</code> for all i). Generated code can use this to optimize this
- * common case. */
+ /**
+ * Advancer for records. The {@link advancers} array contains an advancer for
+ * each field, ordered according writer (which determines the order in which
+ * data must be read). The {@link readerOrder} array tells you how those
+ * advancers line up with the reader's fields. Thus, the following is how to
+ * read a record:
+ *
+ * <pre>
+ * for (int i = 0; i < a.advancers.length; i++) {
+ * dataum.set(a.readerOrder[i].pos(), a.advancers[i].next(in));
+ * }
+ * a.done(in);
+ * </pre>
+ *
+ * Note that a decoder <em>must</em> call {@link done} after interpreting all
+ * the elemnts in {@link advancers}.
+ *
+ * As a convenience, {@link inOrder} is set to true iff the reader and writer
+ * order agrees (i.e., iff <code>readerOrder[i].pos() ==
+ * i</code> for all i). Generated code can use this to optimize this common
+ * case.
+ */
public static class Record extends Advancer {
public final Advancer[] advancers;
private Schema[] finalSkips;
public final Schema.Field[] readerOrder;
public final boolean inOrder;
- private Record(Schema w, Schema r, Advancer[] advancers, Schema[] finalSkips,
- Schema.Field[] readerOrder, boolean inOrder)
- {
- super(w,r);
+ private Record(Schema w, Schema r, Advancer[] advancers, Schema[] finalSkips, Schema.Field[] readerOrder,
+ boolean inOrder) {
+ super(w, r);
this.advancers = advancers;
this.finalSkips = finalSkips;
this.readerOrder = readerOrder;
this.inOrder = inOrder;
}
- public Record getRecordAdvancer(Decoder in) { return this; }
+ public Record getRecordAdvancer(Decoder in) {
+ return this;
+ }
/** Must be called after consuming all elements of {@link advancers}. */
public void done(Decoder in) throws IOException {
@@ -631,7 +897,7 @@ abstract class Advancer {
int nrf = 0; // (Insertion) index into fieldAdvs
// Deal with fields to be read
- for ( ; rf < ra.firstDefault; rf++, nrf++) {
+ for (; rf < ra.firstDefault; rf++, nrf++) {
Schema[] toSkip = collectSkips(ra.fieldActions, i);
i += toSkip.length;
Advancer fieldAdv = Advancer.from(ra.fieldActions[i++]);
@@ -653,8 +919,8 @@ abstract class Advancer {
// decoders to have an optimized path for the common case of a
// record's field order not changing.
boolean inOrder = true;
- for (int k = 0; k < ra.firstDefault-1; k++)
- inOrder &= (readOrder[k].pos() < readOrder[k+1].pos());
+ for (int k = 0; k < ra.firstDefault - 1; k++)
+ inOrder &= (readOrder[k].pos() < readOrder[k + 1].pos());
if (inOrder) {
Advancer[] newAdvancers = new Advancer[fieldAdvs.length];
Schema.Field[] newReadOrder = new Schema.Field[fieldAdvs.length];
@@ -678,79 +944,140 @@ abstract class Advancer {
private static class RecordField extends Advancer {
private final Schema[] toSkip;
private final Advancer field;
+
public RecordField(Schema w, Schema r, Schema[] toSkip, Advancer field) {
- super(w,r);
+ super(w, r);
this.toSkip = toSkip;
this.field = field;
}
- public Object nextNull(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextNull(in); }
+ public Object nextNull(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextNull(in);
+ }
- public boolean nextBoolean(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextBoolean(in); }
+ public boolean nextBoolean(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextBoolean(in);
+ }
- public int nextInt(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextInt(in); }
+ public int nextInt(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextInt(in);
+ }
- public long nextLong(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextLong(in); }
+ public long nextLong(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextLong(in);
+ }
- public float nextFloat(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextFloat(in); }
+ public float nextFloat(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextFloat(in);
+ }
- public double nextDouble(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextDouble(in); }
+ public double nextDouble(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextDouble(in);
+ }
- public int nextEnum(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextEnum(in); }
+ public int nextEnum(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextEnum(in);
+ }
- public String nextString(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextString(in); }
+ public String nextString(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextString(in);
+ }
- public Utf8 nextString(Decoder in, Utf8 old) throws IOException
- { ignore(toSkip, in); return field.nextString(in, old); }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
+ ignore(toSkip, in);
+ return field.nextString(in, old);
+ }
- public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException
- { ignore(toSkip, in); return field.nextBytes(in, old); }
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ ignore(toSkip, in);
+ return field.nextBytes(in, old);
+ }
- public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException
- { ignore(toSkip, in); return field.nextFixed(in, bytes, start, len); }
+ public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
+ ignore(toSkip, in);
+ return field.nextFixed(in, bytes, start, len);
+ }
- public int nextIndex(Decoder in) throws IOException
- { ignore(toSkip, in); return field.nextIndex(in); }
+ public int nextIndex(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.nextIndex(in);
+ }
- public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
- { ignore(toSkip, in); return field.getBranchAdvancer(in, branch); }
+ public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
+ ignore(toSkip, in);
+ return field.getBranchAdvancer(in, branch);
+ }
- public Container getArrayAdvancer(Decoder in) throws IOException
- { ignore(toSkip, in); return field.getArrayAdvancer(in); }
+ public Container getArrayAdvancer(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.getArrayAdvancer(in);
+ }
- public Map getMapAdvancer(Decoder in) throws IOException
- { ignore(toSkip, in); return field.getMapAdvancer(in); }
+ public Map getMapAdvancer(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.getMapAdvancer(in);
+ }
- public Record getRecordAdvancer(Decoder in) throws IOException
- { ignore(toSkip, in); return field.getRecordAdvancer(in); }
+ public Record getRecordAdvancer(Decoder in) throws IOException {
+ ignore(toSkip, in);
+ return field.getRecordAdvancer(in);
+ }
}
private static class Default extends Advancer {
protected final Object val;
- private Default(Schema s, Object v) { super(s,s); val = v; }
- public Object next(Decoder in) { return val; }
- public Object nextNull(Decoder in) { return val; }
- public boolean nextBoolean(Decoder in) { return (Boolean) val; }
- public int nextInt(Decoder in) { return (Integer) val; }
- public long nextLong(Decoder in) { return (Long) val; }
- public float nextFloat(Decoder in) { return (Float) val; }
- public double nextDouble(Decoder in) { return (Double) val; }
- public int nextEnum(Decoder in) { return (Integer) val; }
+ private Default(Schema s, Object v) {
+ super(s, s);
+ val = v;
+ }
+
+ public Object next(Decoder in) {
+ return val;
+ }
+
+ public Object nextNull(Decoder in) {
+ return val;
+ }
+
+ public boolean nextBoolean(Decoder in) {
+ return (Boolean) val;
+ }
+
+ public int nextInt(Decoder in) {
+ return (Integer) val;
+ }
+
+ public long nextLong(Decoder in) {
+ return (Long) val;
+ }
+
+ public float nextFloat(Decoder in) {
+ return (Float) val;
+ }
+
+ public double nextDouble(Decoder in) {
+ return (Double) val;
+ }
+
+ public int nextEnum(Decoder in) {
+ return (Integer) val;
+ }
// TODO -- finish for the rest of the types
}
private static void ignore(Schema[] toIgnore, Decoder in) throws IOException {
- for (Schema s: toIgnore) skip(s, in);
+ for (Schema s : toIgnore)
+ skip(s, in);
}
// Probably belongs someplace else, although Decoder doesn't reference
@@ -758,33 +1085,55 @@ abstract class Advancer {
// new dependencies...
public static void skip(Schema s, Decoder in) throws IOException {
switch (s.getType()) {
- case NULL: in.readNull(); break;
- case BOOLEAN: in.readBoolean(); break;
- case INT: in.readInt(); break;
- case LONG: in.readLong(); break;
- case FLOAT: in.readFloat(); break;
- case DOUBLE: in.readDouble(); break;
- case STRING: in.skipString(); break;
- case BYTES: in.skipBytes(); break;
- case FIXED: in.skipFixed(s.getFixedSize()); break;
- case ENUM: in.readEnum(); break;
- case UNION: skip(s.getTypes().get(in.readInt()), in); break;
+ case NULL:
+ in.readNull();
+ break;
+ case BOOLEAN:
+ in.readBoolean();
+ break;
+ case INT:
+ in.readInt();
+ break;
+ case LONG:
+ in.readLong();
+ break;
+ case FLOAT:
+ in.readFloat();
+ break;
+ case DOUBLE:
+ in.readDouble();
+ break;
+ case STRING:
+ in.skipString();
+ break;
+ case BYTES:
+ in.skipBytes();
+ break;
+ case FIXED:
+ in.skipFixed(s.getFixedSize());
+ break;
+ case ENUM:
+ in.readEnum();
+ break;
+ case UNION:
+ skip(s.getTypes().get(in.readInt()), in);
+ break;
case RECORD:
- for (Schema.Field f: s.getFields())
+ for (Schema.Field f : s.getFields())
skip(f.schema(), in);
break;
case ARRAY:
for (long i = in.skipArray(); i != 0; i = in.skipArray())
- for (long j = 0; j < i; j++)
- skip(s.getElementType(), in);
- break;
+ for (long j = 0; j < i; j++)
+ skip(s.getElementType(), in);
+ break;
case MAP:
for (long k = in.skipArray(); k != 0; k = in.skipArray())
- for (long l = 0; l < k; l++) {
- in.skipString(); // Key
- skip(s.getValueType(), in);
+ for (long l = 0; l < k; l++) {
+ in.skipString(); // Key
+ skip(s.getValueType(), in);
}
- break;
+ break;
default:
throw new IllegalArgumentException("Unknown type for schema: " + s);
}
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
index cbfcff7..5ff9e50 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -37,32 +37,39 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
data = d;
}
- /** ... Document how we use <code>d:</code> to create fixed, array,
- * map, and record objects.
- */
+ /**
+ * ... Document how we use <code>d:</code> to create fixed, array, map, and
+ * record objects.
+ */
public static GenericDatumReader2 getReaderFor(Schema writer, Schema reader, GenericData d) {
// TODO: add caching
Resolver.Action a = Resolver.resolve(writer, reader, d);
- Advancer.Record r = (Advancer.Record)Advancer.from(a);
+ Advancer.Record r = (Advancer.Record) Advancer.from(a);
return new GenericDatumReader2(r, d);
}
public D read(D reuse, Decoder in) throws IOException {
- return null;
+ return null;
}
- public Object read(Object reuse, Advancer a, Decoder in)
- throws IOException
- {
+ public Object read(Object reuse, Advancer a, Decoder in) throws IOException {
switch (a.reader.getType()) {
- case NULL: return a.nextNull(in);
- case BOOLEAN: return (Boolean) a.nextBoolean(in);
- case INT: return (Integer) a.nextInt(in);
- case LONG: return (Long) a.nextLong(in);
- case FLOAT: return (Float) a.nextFloat(in);
- case DOUBLE: return (Double) a.nextDouble(in);
- case STRING: return (String) a.nextString(in);
- case BYTES: return a.nextBytes(in, (ByteBuffer)reuse);
+ case NULL:
+ return a.nextNull(in);
+ case BOOLEAN:
+ return (Boolean) a.nextBoolean(in);
+ case INT:
+ return (Integer) a.nextInt(in);
+ case LONG:
+ return (Long) a.nextLong(in);
+ case FLOAT:
+ return (Float) a.nextFloat(in);
+ case DOUBLE:
+ return (Double) a.nextDouble(in);
+ case STRING:
+ return (String) a.nextString(in);
+ case BYTES:
+ return a.nextBytes(in, (ByteBuffer) reuse);
case FIXED: {
GenericFixed fixed = (GenericFixed) data.createFixed(reuse, a.reader);
a.nextFixed(in, fixed.bytes());
@@ -77,10 +84,11 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
((GenericArray) reuse).reset();
} else if (reuse instanceof Collection) {
((Collection) reuse).clear();
- } else reuse = new GenericData.Array((int)i, a.reader);
+ } else
+ reuse = new GenericData.Array((int) i, a.reader);
- Collection array = (Collection)reuse;
- for( ; i != 0; i = c.nextChunk(in))
+ Collection array = (Collection) reuse;
+ for (; i != 0; i = c.nextChunk(in))
for (long j = 0; j < i; j++) {
Object v = read(null, ec, in);
// TODO -- logical type conversion
@@ -89,7 +97,7 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
if (array instanceof GenericArray<?>)
((GenericArray<?>) array).prune();
}
-
+
case MAP: {
Advancer.Map c = advancer.getMapAdvancer(in);
Advancer kc = c.keyAdvancer;
@@ -97,9 +105,10 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
long i = c.firstChunk(in);
if (reuse instanceof Map) {
((Map) reuse).clear();
- } else reuse = new HashMap<Object,Object>((int)i);
- Map map = (Map)reuse;
- for ( ; i != 0; i = c.nextChunk(in))
+ } else
+ reuse = new HashMap<Object, Object>((int) i);
+ Map map = (Map) reuse;
+ for (; i != 0; i = c.nextChunk(in))
for (int j = 0; j < i; j++) {
Object key = kc.nextString(in);
Object val = read(null, ec, in);
@@ -113,7 +122,7 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
Object r = data.newRecord(reuse, ra.reader);
for (int i = 0; i < ra.advancers.length; i++) {
int p = ra.readerOrder[i].pos();
- ((IndexedRecord)reuse).put(p, read(null, ra.advancers[i], in));
+ ((IndexedRecord) reuse).put(p, read(null, ra.advancers[i], in));
}
ra.done(in);
return r;
@@ -127,8 +136,10 @@ public class GenericDatumReader2<D> implements DatumReader<D> {
}
}
- /** Throws {@link UnsupportedOperationException}. (In retrospect, making
- * DatumReaders mutable wasn't a good idea...) */
+ /**
+ * Throws {@link UnsupportedOperationException}. (In retrospect, making
+ * DatumReaders mutable wasn't a good idea...)
+ */
public void setSchema(Schema s) {
throw new UnsupportedOperationException();
}
[avro] 11/14: Fleshed out rest of cases in GenericReader2 (but do
not handle logical types).
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 5e50f64995559c647fc98e3f87e1aebccec3f7c7
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 16:34:39 2019 -0700
Fleshed out rest of cases in GenericReader2 (but do not handle logical types).
---
.../java/org/apache/avro/generic/Advancer.java | 107 ++++++++++++++-------
.../apache/avro/generic/GenericDatumReader2.java | 96 ++++++++++++------
2 files changed, 137 insertions(+), 66 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index d80e812..7c469cb 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -57,9 +57,9 @@ import org.apache.avro.util.Utf8;
* and return the actual value.)
*
* Traversing records, arrays, and maps is more involved. In the
- * case of an array or map, call {@link getContainerAdvancer} and
- * proceed as described in the documentation for {@link
- * Advancer.Container}. For records, best to just look at the
+ * case of an array or map, call {@link getArrayAdvancer} {@link
+ * getMapAdvancer} and proceed as described in the documentation for
+ * {@link Advancer.Container}. For records, best to just look at the
* implementation of {@link GenericDatumReader2}.
**/
abstract class Advancer {
@@ -112,11 +112,18 @@ abstract class Advancer {
return null;
}
- /** Access to advancer for array or map type. */
- public Container getContainerAdvancer(Decoder in) throws IOException {
+ /** Access to advancer for array type. */
+ public Container getArrayAdvancer(Decoder in) throws IOException {
exception();
return null;
}
+
+ /** Access to advancer for array type. */
+ public Map getMapAdvancer(Decoder in) throws IOException {
+ exception();
+ return null;
+ }
+
/** Access to advancer for record type. */
public Record getRecordAdvancer(Decoder in) throws IOException {
exception();
@@ -168,8 +175,8 @@ abstract class Advancer {
case CONTAINER:
Advancer ea = Advancer.from(((Resolver.Container)a).elementAction);
if (a.writer.getType() == Schema.Type.ARRAY)
- return new ArrayContainer(a.writer, a.reader, ea);
- else return new MapContainer(a.writer, a.reader, ea);
+ return new Container(a.writer, a.reader, ea);
+ else return new Map(a.writer, a.reader, ea);
case RECORD:
return Advancer.Record.from((Resolver.RecordAdjust)a);
@@ -188,7 +195,7 @@ abstract class Advancer {
ru.firstMatch, Advancer.from(ru.actualAction));
case ERROR:
- return new Error(w,r, a.toString());
+ return new Error(a.writer,a.reader, a.toString());
case SKIP:
throw new RuntimeException("Internal error. Skip should've been consumed.");
default:
@@ -224,42 +231,54 @@ abstract class Advancer {
}
}
- /** Used for Array and Map. The following fragment illustrates how
+ /** Used for Array. The following fragment illustrates how
* to use to read an array of int:
*
* <pre>
- * Advancer c = advancer.getContainerAdvancer(in);
- * Advancer.Container ec = c.getElementAdvancer(in);
+ * Advancer.Container c = advancer.getArrayAdvancer(in);
* for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
* for (long j = 0; j < i; j++) {
- * int element = c.readInt(in);
+ * int element = c.elementAdvancer.readInt(in);
* // .. do something with this element
* }
* }
* </pre>
* See the implementation of {@link GenericDatumReader2} for more
* illustrations. */
- public abstract static class Container extends Advancer {
- private final Advancer elementAdvancer;
- public Container(Schema w, Schema r, Advancer ea) { super(wr); elementAdvancer = ea; }
- public Container getContainerAdvancer(Decoder in) { return this; }
- public Advancer getElementAdvancer(Decoder in) { return elementAdvancer; }
- public abstract long firstChunk(Decoder in) throws IOException;
- public abstract long nextChunk(Decoder in) throws IOException;
- }
+ public static class Container extends Advancer {
+ public final Advancer elementAdvancer;
+ public Container(Schema w, Schema r, Advancer ea)
+ { super(w,r); elementAdvancer = ea; }
- private static class ArrayContainer extends Container {
- public ArrayContainer(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
public long firstChunk(Decoder in) throws IOException
{ return in.readArrayStart(); }
+
public long nextChunk(Decoder in) throws IOException
{ return in.arrayNext(); }
}
- private static class MapContainer extends Container {
- public MapContainer(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
+ /** Used for Map. The following fragment illustrates how
+ * to use to read an array of int:
+ *
+ * <pre>
+ * Advancer.Map c = advancer.getMapAdvancer(in);
+ * for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ * for (long j = 0; j < i; j++) {
+ * String key = c.keyAdvancer.readString(in);
+ * int element = c.elementAdvancer.readInt(in);
+ * // .. do something with this element
+ * }
+ * }
+ * </pre>
+ * See the implementation of {@link GenericDatumReader2} for more
+ * illustrations. */
+ public static class Map extends Container {
+ public final Advancer keyAdvancer = StringFast.instance;
+ public Map(Schema w, Schema r, Advancer ea) { super(w,r,ea); }
+
public long firstChunk(Decoder in) throws IOException
{ return in.readMapStart(); }
+
public long nextChunk(Decoder in) throws IOException
{ return in.mapNext(); }
}
@@ -270,7 +289,8 @@ abstract class Advancer {
private static class NullFast extends Advancer {
public static final NullFast instance = new NullFast();
- private NullFast() { Schema s = Schema.create(Schema.Type.NULL); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.NULL);
+ private NullFast() { super(s,s); }
public Object nextNull(Decoder in) throws IOException {
in.readNull();
return null;
@@ -280,7 +300,8 @@ abstract class Advancer {
private static class BooleanFast extends Advancer {
public static final BooleanFast instance = new BooleanFast();
- private BooleanFast() { Schema s = Schema.create(Schema.Type.BOOLEAN); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.BOOLEAN);
+ private BooleanFast() { super(s,s); }
public boolean nextBoolean(Decoder in) throws IOException {
return in.readBoolean();
}
@@ -289,7 +310,8 @@ abstract class Advancer {
private static class IntFast extends Advancer {
public static final IntFast instance = new IntFast();
- private IntFast() { Schema s = Schema.create(Schema.Type.INTEGER); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.INT);
+ private IntFast() { super(s,s); }
public int nextInt(Decoder in) throws IOException {
return in.readInt();
}
@@ -298,7 +320,8 @@ abstract class Advancer {
private static class LongFast extends Advancer {
public static final LongFast instance = new LongFast();
- private LongFast() { Schema s = Schema.create(Schema.Type.LONG); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.LONG);
+ private LongFast() { super(s,s); }
public long nextLong(Decoder in) throws IOException {
return in.readLong();
}
@@ -307,7 +330,8 @@ abstract class Advancer {
private static class FloatFast extends Advancer {
public static final FloatFast instance = new FloatFast();
- private FloatFast() { Schema s = Schema.create(Schema.Type.FLOAT); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.FLOAT);
+ private FloatFast() { super(s,s); }
public float nextFloat(Decoder in) throws IOException {
return in.readFloat();
}
@@ -316,7 +340,8 @@ abstract class Advancer {
private static class DoubleFast extends Advancer {
public static final DoubleFast instance = new DoubleFast();
- private DoubleFast() { Schema s = Schema.create(Schema.Type.DOUBLE); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.DOUBLE);
+ private DoubleFast() { super(s,s); }
public double nextDouble(Decoder in) throws IOException {
return in.readDouble();
}
@@ -325,7 +350,8 @@ abstract class Advancer {
private static class StringFast extends Advancer {
public static final StringFast instance = new StringFast();
- private StringFast() { Schema s = Schema.create(Schema.Type.STRING); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.STRING);
+ private StringFast() { super(s,s); }
public String nextString(Decoder in) throws IOException { return in.readString(); }
public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
return in.readString(old);
@@ -335,7 +361,8 @@ abstract class Advancer {
private static class BytesFast extends Advancer {
public static final BytesFast instance = new BytesFast();
- private BytesFast() { Schema s = Schema.create(Schema.Type.BYTES); super(s,s); }
+ private static final Schema s = Schema.create(Schema.Type.BYTES);
+ private BytesFast() { super(s,s); }
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
return in.readBytes(old);
}
@@ -477,7 +504,7 @@ abstract class Advancer {
* consume the tag ourself and call the corresponding advancer. */
private static class WriterUnion extends Advancer {
private Advancer[] branches;
- public WriterUnion(Schema w, Schema r, Advancer[] b) { super(w,r) branches = b; }
+ public WriterUnion(Schema w, Schema r, Advancer[] b) { super(w,r); branches = b; }
private final Advancer b(Decoder in) throws IOException
{ return branches[in.readIndex()]; }
@@ -504,8 +531,11 @@ abstract class Advancer {
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
{ return b(in).getBranchAdvancer(in, branch); }
- public Container getContainerAdvancer(Decoder in) throws IOException
- { return b(in).getContainerAdvancer(in); }
+ public Container getArrayAdvancer(Decoder in) throws IOException
+ { return b(in).getArrayAdvancer(in); }
+
+ public Map getMapAdvancer(Decoder in) throws IOException
+ { return b(in).getMapAdvancer(in); }
public Record getRecordAdvancer(Decoder in) throws IOException
{ return b(in).getRecordAdvancer(in); }
@@ -693,8 +723,11 @@ abstract class Advancer {
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
{ ignore(toSkip, in); return field.getBranchAdvancer(in, branch); }
- public Container getContainerAdvancer(Decoder in) throws IOException
- { ignore(toSkip, in); return field.getContainerAdvancer(in); }
+ public Container getArrayAdvancer(Decoder in) throws IOException
+ { ignore(toSkip, in); return field.getArrayAdvancer(in); }
+
+ public Map getMapAdvancer(Decoder in) throws IOException
+ { ignore(toSkip, in); return field.getMapAdvancer(in); }
public Record getRecordAdvancer(Decoder in) throws IOException
{ ignore(toSkip, in); return field.getRecordAdvancer(in); }
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
index 46506c2..cbfcff7 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader2.java
@@ -19,47 +19,42 @@ package org.apache.avro.generic;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.List;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.avro.Resolver;
import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
-public class GenericDatumReader2<D extends IndexedRecord> implements DatumReader<D> {
- private final Schema reader, writer;
+public class GenericDatumReader2<D> implements DatumReader<D> {
private final Advancer.Record advancer;
private final GenericData data;
- private GenericDatumReader2(Schema writer, Schema reader, Advancer.Record a, GenericData d) {
- this.writer = writer;
- this.reader = reader;
+ private GenericDatumReader2(Advancer.Record a, GenericData d) {
advancer = a;
data = d;
}
- public static GenericDatumReader2 getReaderFor(Schema writer, Schema reader, GenericData data) {
+ /** ... Document how we use <code>d:</code> to create fixed, array,
+ * map, and record objects.
+ */
+ public static GenericDatumReader2 getReaderFor(Schema writer, Schema reader, GenericData d) {
// TODO: add caching
- Resolver.Action a = Resolver.resolve(writer, reader, data);
+ Resolver.Action a = Resolver.resolve(writer, reader, d);
Advancer.Record r = (Advancer.Record)Advancer.from(a);
- return new GenericDatumReader2(writer, reader, r, data);
+ return new GenericDatumReader2(r, d);
}
public D read(D reuse, Decoder in) throws IOException {
- List<Schema.Field> wf = writer.getFields();
- if (reuse == null) reuse = null; // FIXME
- for (int i = 0; i < advancer.advancers.length; i++) {
- int p = advancer.readerOrder[i].pos();
- reuse.put(p, read(null, wf.get(i).schema(), advancer.advancers[i], in));
- }
- advancer.done(in);
- return reuse;
+ return null;
}
- public Object read(Object reuse, Schema expected, Advancer a, Decoder in)
+ public Object read(Object reuse, Advancer a, Decoder in)
throws IOException
{
- switch (expected.getType()) {
+ switch (a.reader.getType()) {
case NULL: return a.nextNull(in);
case BOOLEAN: return (Boolean) a.nextBoolean(in);
case INT: return (Integer) a.nextInt(in);
@@ -68,22 +63,65 @@ public class GenericDatumReader2<D extends IndexedRecord> implements DatumReader
case DOUBLE: return (Double) a.nextDouble(in);
case STRING: return (String) a.nextString(in);
case BYTES: return a.nextBytes(in, (ByteBuffer)reuse);
- case FIXED:
+ case FIXED: {
+ GenericFixed fixed = (GenericFixed) data.createFixed(reuse, a.reader);
+ a.nextFixed(in, fixed.bytes());
+ return fixed;
+ }
+
case ARRAY: {
- List result = null; // FIXME -- use GenericData methods here...
- Advancer.Container c = advancer.getContainerAdvancer(in);
- Advancer ec = c.getElementAdvancer(in);
- Schema es = expected.getElementType();
- for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ Advancer.Container c = advancer.getArrayAdvancer(in);
+ Advancer ec = c.elementAdvancer;
+ long i = c.firstChunk(in);
+ if (reuse instanceof GenericArray) {
+ ((GenericArray) reuse).reset();
+ } else if (reuse instanceof Collection) {
+ ((Collection) reuse).clear();
+ } else reuse = new GenericData.Array((int)i, a.reader);
+
+ Collection array = (Collection)reuse;
+ for( ; i != 0; i = c.nextChunk(in))
for (long j = 0; j < i; j++) {
- result.add(read(null, es, ec, in));
+ Object v = read(null, ec, in);
+ // TODO -- logical type conversion
+ array.add(v);
}
- }
+ if (array instanceof GenericArray<?>)
+ ((GenericArray<?>) array).prune();
}
- case MAP:
- case RECORD:
+ case MAP: {
+ Advancer.Map c = advancer.getMapAdvancer(in);
+ Advancer kc = c.keyAdvancer;
+ Advancer ec = c.elementAdvancer;
+ long i = c.firstChunk(in);
+ if (reuse instanceof Map) {
+ ((Map) reuse).clear();
+ } else reuse = new HashMap<Object,Object>((int)i);
+ Map map = (Map)reuse;
+ for ( ; i != 0; i = c.nextChunk(in))
+ for (int j = 0; j < i; j++) {
+ Object key = kc.nextString(in);
+ Object val = read(null, ec, in);
+ map.put(key, val);
+ }
+ return map;
+ }
+
+ case RECORD: {
+ Advancer.Record ra = advancer.getRecordAdvancer(in);
+ Object r = data.newRecord(reuse, ra.reader);
+ for (int i = 0; i < ra.advancers.length; i++) {
+ int p = ra.readerOrder[i].pos();
+ ((IndexedRecord)reuse).put(p, read(null, ra.advancers[i], in));
+ }
+ ra.done(in);
+ return r;
+ }
+
case UNION:
+ return read(reuse, advancer.getBranchAdvancer(in, advancer.nextIndex(in)), in);
+
default:
throw new IllegalArgumentException("Can't handle this yet.");
}
[avro] 01/14: Small changes to Resolver to support Advancer
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 0ec4b0c9c598ca3c4692a984b50ee539bb630b69
Author: raymie_stata <ra...@raymiestatasMBP.lan>
AuthorDate: Mon Apr 29 21:49:03 2019 -0700
Small changes to Resolver to support Advancer
---
.../src/main/java/org/apache/avro/Resolver.java | 80 +++++++++++-----------
1 file changed, 40 insertions(+), 40 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
index 3dab690..7c9e2c3 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/Resolver.java
@@ -163,10 +163,11 @@ public class Resolver {
}
/**
- * In this case, there's nothing to be done for resolution: the two schemas are
- * effectively the same. This action will be generated <em>only</em> for
- * primitive types and fixed types, and not for any other kind of schema.
- */
+ * In this case, there's nothing to be done for resolution: the two
+ * schemas are effectively the same. This action will be generated
+ * <em>only</em> for primitive types (incl. STRING and BYTES) and
+ * also for fixed types, and <not>not</not for any other kind of
+ * schema. */
public static class DoNothing extends Action {
public DoNothing(Schema w, Schema r, GenericData d) {
super(w, r, d, Action.Type.DO_NOTHING);
@@ -249,8 +250,18 @@ public class Resolver {
* promotion"), and whent the promotion is one allowed by the Avro spec.
*/
public static class Promote extends Action {
- private Promote(Schema w, Schema r, GenericData d) {
+ public static enum Promotions {
+ INT2LONG, INT2FLOAT, INT2DOUBLE,
+ LONG2FLOAT, LONG2DOUBLE, FLOAT2DOUBLE,
+ STRING2BYTES, BYTES2STRING
+ }
+
+ /** The exact promotion being represented. */
+ public final Promotions promotion;
+
+ private Promote(Schema w, Schema r, Promotions p, GenericData d) {
super(w, r, d, Action.Type.PROMOTE);
+ this.promotion = p;
}
/**
@@ -264,62 +275,49 @@ public class Resolver {
* not different.
*/
public static Action resolve(Schema w, Schema r, GenericData d) {
- if (isValid(w, r))
- return new Promote(w, r, d);
+ Promotions p = findPromotion(w, r);
+ if (p != null)
+ return new Promote(w, r, p, d);
else
return new ErrorAction(w, r, d, ErrorType.INCOMPATIBLE_SCHEMA_TYPES);
}
/**
- * Returns true iff <tt>w</tt> and <tt>r</tt> are both primitive types and
- * either they are the same type or <tt>w</tt> is promotable to <tt>r</tt>.
- * Should
+ * Returns the {@link Promtion} that would resolve schema
+ * <code>w</code> against <code>r</code>, or null if there is no
+ * such promotion.
*/
- public static boolean isValid(Schema w, Schema r) {
+ private static Promotions findPromotion(Schema w, Schema r) {
if (w.getType() == r.getType())
throw new IllegalArgumentException("Only use when reader and writer are different.");
Schema.Type wt = w.getType();
switch (r.getType()) {
- case INT:
- switch (wt) {
- case INT:
- return true;
- }
- break;
case LONG:
switch (wt) {
- case INT:
- case LONG:
- return true;
+ case INT: return Promotions.INT2LONG;
}
break;
case FLOAT:
switch (wt) {
- case INT:
- case LONG:
- case FLOAT:
- return true;
+ case INT: return Promotions.INT2FLOAT;
+ case LONG: return Promotions.LONG2FLOAT;
}
break;
case DOUBLE:
switch (wt) {
- case INT:
- case LONG:
- case FLOAT:
- case DOUBLE:
- return true;
+ case INT: return Promotions.INT2DOUBLE;
+ case LONG: return Promotions.LONG2DOUBLE;
+ case FLOAT: return Promotions.FLOAT2DOUBLE;
}
break;
case BYTES:
+ if (wt == Schema.Type.STRING) return Promotions.STRING2BYTES;
+ break;
case STRING:
- switch (wt) {
- case STRING:
- case BYTES:
- return true;
- }
+ if (wt == Schema.Type.STRING) return Promotions.BYTES2STRING;
break;
}
- return false;
+ return null;
}
}
@@ -424,11 +422,13 @@ public class Resolver {
public final Action[] fieldActions;
/**
- * Contains (all of) the reader's fields. The first <i>n</i> of these are the
- * fields that will be read from the writer: these <i>n</i> are in the order
- * dictated by writer's schema. The remaining <i>m</i> fields will be read from
- * default values (actions for these default values are found in
- * {@link RecordAdjust#defaults}.
+ * Contains (all of) the reader's fields. The first {@link
+ * firstDefault} of these are the fields that will be read from
+ * the writer: these are in the order dictated by writer's
+ * schema. The remaining fields from {@link firstDefault} to the
+ * end of the array will be read from default values (actions for
+ * these default values are found in {@link defaults}. Note that
+ * the default fields are in the order of the reader's schema.
*/
public final Field[] readerOrder;
[avro] 08/14: Changes for what is needed for arrays and maps.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 32999b8bd7872aaf5488a52966b5605cdf98d82d
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 11:05:49 2019 -0700
Changes for what is needed for arrays and maps.
---
.../java/org/apache/avro/generic/Advancer.java | 88 +++++++++++++++-------
1 file changed, 60 insertions(+), 28 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index d75e9ee..4fe3d7b 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -46,10 +46,9 @@ import org.apache.avro.util.Utf8;
*
* When traversing an Array or Map in the reader schema, the decoder
* should call {@link getElementAdvancer} to retrieve the advancer
- * object for the contained element-schema, value-schema, or non-null
- * schema respectively. ({@link next} cannot be called on {@link
- * Advancer.Record} objects -- decoders must decode them field by
- * field.)
+ * object for the contained element-schema or value-schema. See the
+ * JavaDoc for {@link getElementAdvancer} for instructions on how to
+ * decode these types.
*
* For unions, the decoder should call {@link nextIndex} to fetch the
* branch and then {@link getBranchAdvancer} to get the advancer of
@@ -57,11 +56,11 @@ import org.apache.avro.util.Utf8;
* index, pick the right advancer based on the index, and then read
* and return the actual value.)
*
- * Traversing an record is more involved. The decoder should call
- * {@link getRecordAdvancer} and proceed as described in the
- * documentation for {@link Advancer.Record}. ({@link next} cannot
- * be called on {@link Advancer.Record} objects -- decoders must
- * decode them field by field.)
+ * Traversing records, arrays, and maps is more involved. In the
+ * case of an array or map, call {@link getContainerAdvancer} and
+ * proceed as described in the documentation for {@link
+ * Advancer.Container}. For records, best to just look at the
+ * implementation of {@link GenericDatumReader2}.
**/
abstract class Advancer {
protected Exception exception() {
@@ -100,12 +99,6 @@ abstract class Advancer {
return nextFixed(in, bytes, 0, bytes.length);
}
- /** Access to contained advancer (for Array and Map types). */
- public Advancer getElementAdvancer(Decoder in) throws IOException {
- exception();
- return null;
- }
-
/** Get index for a union. */
public int nextIndex(Decoder in) throws IOException { exception(); return 0; }
@@ -116,7 +109,12 @@ abstract class Advancer {
return null;
}
- /** Access to contained advancer (for Array, Map, and Union types). */
+ /** Access to advancer for array or map type. */
+ public Container getContainerAdvancer(Decoder in) throws IOException {
+ exception();
+ return null;
+ }
+ /** Access to advancer for record type. */
public Record getRecordAdvancer(Decoder in) throws IOException {
exception();
return null;
@@ -165,7 +163,9 @@ abstract class Advancer {
else return new EnumWithAdjustments(e.adjustments);
case CONTAINER:
- return new Container(Advancer.from(((Resolver.Container)a).elementAction));
+ Advancer ea = Advancer.from(((Resolver.Container)a).elementAction);
+ if (a.writer.getType() == Schema.Type.ARRAY) return new ArrayContainer(ea);
+ else return new MapContainer(ea);
case RECORD:
return Advancer.Record.from((Resolver.RecordAdjust)a);
@@ -219,14 +219,46 @@ abstract class Advancer {
}
}
- /** Used for Array, Map, and Union. In case of Union, since we only
- * support "nullable" unions (ie, two-branch unions in which one
- * branch is null), the element advancer is for the non-null branch
- * of the union. */
- private static class Container extends Advancer {
+ /** Used for Array and Map. The following fragment illustrates how
+ * to use to read an array of int:
+ *
+ * <pre>
+ * Advancer.Container c = advancer.getContainerAdvancer(in);
+ * Advancer.Container ec = c.getElementAdvancer(in);
+ * for(long i = c.firstChunk(in); i != 0; i = c.nextChunk(in)) {
+ * for (long j = 0; j < i; j++) {
+ * int element = c.readInt(in);
+ * // .. do something with this element
+ * }
+ * }
+ * </pre>
+ * See the implementation of {@link GenericDatumReader2} for more
+ * illustrations. */
+ public abstract static class Container extends Advancer {
private final Advancer elementAdvancer;
public Container(Advancer elementAdvancer) { this.elementAdvancer = elementAdvancer; }
+ public Container getContainerAdvancer(Decoder in) { return this; }
public Advancer getElementAdvancer(Decoder in) { return elementAdvancer; }
+ public abstract long firstChunk(Decoder in) throws IOException;
+ public abstract long nextChunk(Decoder in) throws IOException;
+ }
+
+ private static class ArrayContainer extends Container {
+ private final Advancer elementAdvancer;
+ public ArrayContainer(Advancer elementAdvancer) { super(elementAdvancer); }
+ public long firstChunk(Decoder in) throws IOException
+ { return in.readArrayStart(); }
+ public long nextChunk(Decoder in) throws IOException
+ { return in.arrayNext(); }
+ }
+
+ private static class MapContainer extends Container {
+ private final Advancer elementAdvancer;
+ public MapContainer(Advancer elementAdvancer) { super(elementAdvancer); }
+ public long firstChunk(Decoder in) throws IOException
+ { return in.readMapStart(); }
+ public long nextChunk(Decoder in) throws IOException
+ { return in.mapNext(); }
}
//// The following set of subclasses are for when there is no
@@ -449,13 +481,13 @@ abstract class Advancer {
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException
{ return b(in).nextFixed(in, bytes, start, length); }
- public Advancer getElementAdvancer(Decoder in) throws IOException
- { return b(in).getElementAdvancer(in); }
-
public int nextIndex(Decoder in) throws IOException { return b(in).nextIndex(in); }
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
{ return b(in).getBranchAdvancer(in, branch); }
+ public Container getContainerAdvancer(Decoder in) throws IOException
+ { return b(in).getContainerAdvancer(in); }
+
public Record getRecordAdvancer(Decoder in) throws IOException
{ return b(in).getRecordAdvancer(in); }
}
@@ -630,15 +662,15 @@ abstract class Advancer {
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException
{ ignore(toSkip, in); return field.nextFixed(in, bytes, start, len); }
- public Advancer getElementAdvancer(Decoder in) throws IOException
- { ignore(toSkip, in); return field.getElementAdvancer(in); }
-
public int nextIndex(Decoder in) throws IOException
{ ignore(toSkip, in); return field.nextIndex(in); }
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
{ ignore(toSkip, in); return field.getBranchAdvancer(in, branch); }
+ public Container getContainerAdvancer(Decoder in) throws IOException
+ { ignore(toSkip, in); return field.getContainerAdvancer(in); }
+
public Record getRecordAdvancer(Decoder in) throws IOException
{ ignore(toSkip, in); return field.getRecordAdvancer(in); }
}
[avro] 05/14: Fixups from previous two commits.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit a22c22b8f2092e19e91b0acf9ddc0a4875ff93ce
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 00:38:40 2019 -0700
Fixups from previous two commits.
---
.../java/org/apache/avro/generic/Advancer.java | 119 +++++----------------
1 file changed, 25 insertions(+), 94 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
index ee326e2..9064a39 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/generic/Advancer.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.avro.specific;
+package org.apache.avro.generic;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -126,8 +126,10 @@ abstract class Advancer {
////// Here's the builder for Advancer trees. The subclasses used by
////// this implementation are found below.
- /** Build an {@link Advancer} tree that for a given {@link
- * Resolver.Action} tree. */
+ /** Build a {@link Advancer} tree that for a given {@link
+ * Resolver.Action} tree. If input argument (<code>a</code>) is a
+ * {@link Resolver.RecordAdjust}, the result is guaranteed to be a
+ * {@link Advancer.Record}. */
public static Advancer from(Resolver.Action a) {
switch (a.type) {
case DO_NOTHING:
@@ -553,7 +555,7 @@ abstract class Advancer {
Schema[] toSkip = collectSkips(ra.fieldActions, i);
i += toSkip.length;
Advancer fieldAdv = Advancer.from(ra.fieldActions[i++]);
- if (toSkip.length != 0) fieldAdv = new RecordField(fieldAdv, toSkip);
+ if (toSkip.length != 0) fieldAdv = new RecordField(toSkip, fieldAdv);
fieldAdvs[nrf] = fieldAdv;
}
@@ -587,133 +589,62 @@ abstract class Advancer {
}
private static class RecordField extends Advancer {
+ private final Schema[] toSkip;
private final Advancer field;
- private final Schema[] after;
- public RecordField(Advancer field, Schema[] after) {
+ public RecordField(Schema[] toSkip, Advancer field) {
+ this.toSkip = toSkip;
this.field = field;
- this.after = after;
}
public Object next(Decoder in) throws IOException
- { Object r = field.next(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.next(in); }
public Object nextNull(Decoder in) throws IOException
- { field.nextNull(in); ignore(after, in); return null; }
+ { ignore(toSkip, in); return field.nextNull(in); }
public boolean nextBoolean(Decoder in) throws IOException
- { boolean r = field.nextBoolean(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextBoolean(in); }
public int nextInt(Decoder in) throws IOException
- { int r = field.nextInt(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextInt(in); }
public long nextLong(Decoder in) throws IOException
- { long r = field.nextLong(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextLong(in); }
public float nextFloat(Decoder in) throws IOException
- { float r = field.nextFloat(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextFloat(in); }
public double nextDouble(Decoder in) throws IOException
- { double r = field.nextDouble(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextDouble(in); }
public int nextEnum(Decoder in) throws IOException
- { int r = field.nextEnum(in); ignore(after, in); return r; }
+ { ignore(toSkip, in); return field.nextEnum(in); }
public String nextString(Decoder in) throws IOException
- { String r = field.nextString(in); ignore(after, in); return r; }
-
- public Utf8 nextString(Decoder in, Utf8 old) throws IOException {
- Utf8 r = field.nextString(in,old);
- ignore(after, in);
- return r;
- }
-
- public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
- ByteBuffer r = field.nextBytes(in,old);
- ignore(after, in);
- return r;
- }
-
- public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException {
- byte[] r = field.nextFixed(in, bytes, start, len);
- ignore(after, in);
- return r;
- }
-
- // TODO: THIS DOESN'T WORK!!
- public Advancer getElementAdvancer(Decoder in) throws IOException
- { Advancer r = field.getElementAdvancer(in); ignore(after, in); return r; }
-
- // TODO: THIS DOESN'T WORK!!
- public int nextIndex(Decoder in) throws IOException
- { int r = field.nextIndex(in); ignore(after, in); return r; }
-
- // TODO: THIS DOESN'T WORK!!
- public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
- { Advancer r = field.getBranchAdvancer(in, branch); ignore(after, in); return r; }
-
- // TODO: THIS DOESN'T WORK!!
- public Record getRecordAdvancer(Decoder in) throws IOException
- { Record r = field.getRecordAdvancer(in); ignore(after, in); return r; }
- }
-
- private static class RecordFieldWithBefore extends Advancer {
- private final Schema[] before;
- private final Advancer field;
- public RecordFieldWithBefore(Schema[] before, Advancer field) {
- this.before = before;
- this.field = field;
- }
-
- public Object next(Decoder in) throws IOException
- { ignore(before, in); return field.next(in); }
-
- public Object nextNull(Decoder in) throws IOException
- { ignore(before, in); return field.nextNull(in); }
-
- public boolean nextBoolean(Decoder in) throws IOException
- { ignore(before, in); return field.nextBoolean(in); }
-
- public int nextInt(Decoder in) throws IOException
- { ignore(before, in); return field.nextInt(in); }
-
- public long nextLong(Decoder in) throws IOException
- { ignore(before, in); return field.nextLong(in); }
-
- public float nextFloat(Decoder in) throws IOException
- { ignore(before, in); return field.nextFloat(in); }
-
- public double nextDouble(Decoder in) throws IOException
- { ignore(before, in); return field.nextDouble(in); }
-
- public int nextEnum(Decoder in) throws IOException
- { ignore(before, in); return field.nextEnum(in); }
-
- public String nextString(Decoder in) throws IOException
- { ignore(before, in); return field.nextString(in); }
+ { ignore(toSkip, in); return field.nextString(in); }
public Utf8 nextString(Decoder in, Utf8 old) throws IOException
- { ignore(before, in); return field.nextString(in, old); }
+ { ignore(toSkip, in); return field.nextString(in, old); }
public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException
- { ignore(before, in); return field.nextBytes(in, old); }
+ { ignore(toSkip, in); return field.nextBytes(in, old); }
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int len) throws IOException
- { ignore(before, in); return field.nextFixed(in, bytes, start, len); }
+ { ignore(toSkip, in); return field.nextFixed(in, bytes, start, len); }
public Advancer getElementAdvancer(Decoder in) throws IOException
- { ignore(before, in); return field.getElementAdvancer(in); }
+ { ignore(toSkip, in); return field.getElementAdvancer(in); }
public int nextIndex(Decoder in) throws IOException
- { ignore(before, in); return field.nextIndex(in); }
+ { ignore(toSkip, in); return field.nextIndex(in); }
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException
- { ignore(before, in); return field.getBranchAdvancer(in, branch); }
+ { ignore(toSkip, in); return field.getBranchAdvancer(in, branch); }
public Record getRecordAdvancer(Decoder in) throws IOException
- { ignore(before, in); return field.getRecordAdvancer(in); }
+ { ignore(toSkip, in); return field.getRecordAdvancer(in); }
}
-
private static class Default extends Advancer {
protected final Object val;
private Default(Object val) { this.val = val; }
[avro] 03/14: Dealt with record field-skipping problem. Also fixed
a few errors missed earlier.
Posted by th...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
thiru pushed a commit to branch fast-decoder-thiru
in repository https://gitbox.apache.org/repos/asf/avro.git
commit 345f6b32f9cc9de75cf492785a62e8a91c78085e
Author: rstata <rs...@yahoo.com>
AuthorDate: Tue Apr 30 00:20:14 2019 -0700
Dealt with record field-skipping problem. Also fixed a few errors missed earlier.
---
.../java/org/apache/avro/specific/Advancer.java | 96 +++++++++++++---------
1 file changed, 56 insertions(+), 40 deletions(-)
diff --git a/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java b/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
index 75d2615..ee326e2 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/specific/Advancer.java
@@ -76,20 +76,24 @@ abstract class Advancer {
//// an integer is read with no promotion) overrides just
//// readInt.
- public Object next(Decoder in) throws IOException { exception(); }
- public Object nextNull(Decoder in) throws IOException { exception(); }
- public boolean nextBoolean(Decoder in) throws IOException { exception(); }
- public int nextInt(Decoder in) throws IOException { exception(); }
- public long nextLong(Decoder in) throws IOException { exception(); }
- public float nextFloat(Decoder in) throws IOException { exception(); }
- public double nextDouble(Decoder in) throws IOException { exception(); }
- public int nextEnum(Decoder in) throws IOException { exception(); }
- public Utf8 nextString(Decoder in, Utf8 old) throws IOException { exception(); }
- public String nextString(Decoder in) throws IOException { exception(); }
- public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException { exception(); }
+ public Object next(Decoder in) throws IOException { exception(); return null; }
+ public Object nextNull(Decoder in) throws IOException { exception(); return null; }
+ public boolean nextBoolean(Decoder in) throws IOException { exception(); return false; }
+ public int nextInt(Decoder in) throws IOException { exception(); return 0; }
+ public long nextLong(Decoder in) throws IOException { exception(); return 0; }
+ public float nextFloat(Decoder in) throws IOException { exception(); return 0; }
+ public double nextDouble(Decoder in) throws IOException { exception(); return 0; }
+ public int nextEnum(Decoder in) throws IOException { exception(); return 0; }
+ public Utf8 nextString(Decoder in, Utf8 old) throws IOException { exception(); return null; }
+ public String nextString(Decoder in) throws IOException { exception(); return null; }
+ public ByteBuffer nextBytes(Decoder in, ByteBuffer old) throws IOException {
+ exception();
+ return null;
+ }
public byte[] nextFixed(Decoder in, byte[] bytes, int start, int length) throws IOException {
exception();
+ return null;
}
public byte[] nextFixed(Decoder in, byte[] bytes) throws IOException {
@@ -99,20 +103,23 @@ abstract class Advancer {
/** Access to contained advancer (for Array and Map types). */
public Advancer getElementAdvancer(Decoder in) throws IOException {
exception();
+ return null;
}
/** Get index for a union. */
- public int nextIndex(Decoder in) throws IOException { exception(); }
+ public int nextIndex(Decoder in) throws IOException { exception(); return 0; }
/** Access to contained advancer for unions. You must call {@link
* nextIndex} before calling this method. */
public Advancer getBranchAdvancer(Decoder in, int branch) throws IOException {
exception();
+ return null;
}
/** Access to contained advancer (for Array, Map, and Union types). */
public Record getRecordAdvancer(Decoder in) throws IOException {
exception();
+ return null;
}
@@ -216,6 +223,7 @@ abstract class Advancer {
* of the union. */
private static class Container extends Advancer {
private final Advancer elementAdvancer;
+ public Container(Advancer elementAdvancer) { this.elementAdvancer = elementAdvancer; }
public Advancer getElementAdvancer(Decoder in) { return elementAdvancer; }
}
@@ -486,32 +494,44 @@ abstract class Advancer {
//// Records are particularly intricate because we may have to skip
//// fields, read fields out of order, and use default values.
- /** Advancer for records. The {@link advancer} array contains an
+ /** Advancer for records. The {@link advancers} array contains an
* advancer for each field, ordered according writer (which
* determines the order in which data must be read). The {@link
* readerOrder} array tells you how those advancers line up with the
* reader's fields. Thus, the following is how to read a record:
* <pre>
- * for (int i = 0; i < a.advancers.length; i++)
+ * for (int i = 0; i < a.advancers.length; i++) {
* dataum.set(a.readerOrder[i], a.advancers[i].next());
+ * }
+ * a.done();
* </pre>
+ * Note that a decoder <em>must</em> call {@link done} after interpreting
+ * all the elemnts in {@link advancers}.
+ *
* As a convenience, {@link inOrder} is set to true iff the reader
* and writer order agrees (i.e., iff <code>readerOrder[i] ==
* i</code> for all i). Generated code can use this to optimize this
* common case. */
public static class Record extends Advancer {
public final Advancer[] advancers;
+ private Schema[] finalSkips;
public final int[] readerOrder;
public final boolean inOrder;
- private Record(Advancer[] advancers, int[] readerOrder, boolean inOrder) {
+ private Record(Advancer[] advancers, Schema[] finalSkips, int[] order, boolean inOrder) {
this.advancers = advancers;
- this.readerOrder = readerOrder;
+ this.finalSkips = finalSkips;
+ this.readerOrder = order;
this.inOrder = inOrder;
}
public Record getRecordAdvancer(Decoder in) { return this; }
+ /** Must be called after consuming all elements of {@link advancers}. */
+ public void done(Decoder in) throws IOException {
+ ignore(finalSkips, in);
+ }
+
protected static Advancer from(Resolver.RecordAdjust ra) {
/** Two cases: reader + writer agree on order, vs disagree. */
/** This is the complicated case, since skipping is involved. */
@@ -525,48 +545,44 @@ abstract class Advancer {
Advancer[] fieldAdvs = new Advancer[readOrder.length];
int i = 0; // Index into ra.fieldActions
- int rf = 0; // Index into readOrder
- int nrf = 0; // Index into fieldAdvs
-
- // Deal with any leading fields to be skipped
- Schema[] firstSkips = collectSkips(ra.fieldActions, i);
- if (firstSkips.length != 0) i += firstSkips.length;
- else firstSkips = null;
+ int rf = 0; // Index into ra.readerOrder
+ int nrf = 0; // (Insertion) index into fieldAdvs
// Deal with fields to be read
- for ( ; i < ra.fieldActions.length; nrf++, rf++) {
- Advancer fieldAdv = Advancer.from(ra.fieldActions[i]);
- i++;
+ for ( ; rf < ra.firstDefault; rf++, nrf++) {
Schema[] toSkip = collectSkips(ra.fieldActions, i);
- if (toSkip.length != 0) {
- fieldAdv = new RecordField(fieldAdv, toSkip);
- i += toSkip.length;
- }
- if (firstSkips != null) {
- fieldAdv = new RecordFieldWithBefore(firstSkips, fieldAdv);
- firstSkips = null;
- }
+ i += toSkip.length;
+ Advancer fieldAdv = Advancer.from(ra.fieldActions[i++]);
+ if (toSkip.length != 0) fieldAdv = new RecordField(fieldAdv, toSkip);
fieldAdvs[nrf] = fieldAdv;
}
+ // Deal with any trailing fields to be skipped:
+ Schema[] finalSkips = collectSkips(ra.fieldActions, i);
+ // Assert i == ra.fieldActions.length
+
+ // Deal with defaults
+ for (int df = 0; rf < readOrder.length; rf++, df++, nrf++)
+ fieldAdvs[nrf] = new Default(ra.defaults[df]);
+
// If reader and writer orders agree, sort fieldAdvs by reader
// order (i.e., move defaults into the correct place), to allow
// decoders to have an optimized path for the common case of a
// record's field order not changing.
boolean inOrder = true;
for (int k = 0; k < ra.firstDefault-1; k++)
- if (readOrder[k] > readOrder[k+1]) inOrder = false;
+ inOrder &= (readOrder[k] < readOrder[k+1]);
if (inOrder) {
Advancer[] newAdvancers = new Advancer[fieldAdvs.length];
for (int k = 0, rf2 = 0, df = ra.firstDefault; k < readOrder.length; k++) {
- if (rf2 < df) newAdvancers[k] = fieldAdvs[rf2++];
- else newAdvancers[k] = fieldAdvs[df++];
- readOrder[k] = k;
+ if (readOrder[rf2] < readOrder[df]) newAdvancers[k] = fieldAdvs[rf2++];
+ else newAdvancers[k] = fieldAdvs[df++];
}
- newAdvancers = fieldAdvs;
+ for (int k = 0; k < readOrder.length; k++) readOrder[k] = k;
+ fieldAdvs = newAdvancers;
}
- return new Record(fieldAdvs, readOrder, inOrder);
+ return new Record(fieldAdvs, finalSkips, readOrder, inOrder);
}
}