add long field support

src/main/java/org/elasticsearch/index/fielddata/IndexFieldDataService.java

@@ -31,6 +31,7 @@ import org.elasticsearch.index.AbstractIndexComponent;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.fielddata.plain.ConcreteBytesRefIndexFieldData;
 import org.elasticsearch.index.fielddata.plain.DoubleArrayIndexFieldData;
+import org.elasticsearch.index.fielddata.plain.LongArrayIndexFieldData;
 import org.elasticsearch.index.settings.IndexSettings;
 
 import java.util.concurrent.ConcurrentMap;
@@ -46,11 +47,13 @@ public class IndexFieldDataService extends AbstractIndexComponent {
         buildersByType = MapBuilder.<String, IndexFieldData.Builder>newMapBuilder()
                 .put("string", new ConcreteBytesRefIndexFieldData.Builder())
                 .put("double", new DoubleArrayIndexFieldData.Builder())
+                .put("long", new LongArrayIndexFieldData.Builder())
                 .immutableMap();
 
-        buildersByTypeAndFormat = MapBuilder.<Tuple<String, String>, IndexFieldData.Builder>newMapBuilder().
+        buildersByTypeAndFormat = MapBuilder.<Tuple<String, String>, IndexFieldData.Builder>newMapBuilder()
-                put(Tuple.tuple("string", "concrete_bytes"), new ConcreteBytesRefIndexFieldData.Builder())
+                .put(Tuple.tuple("string", "concrete_bytes"), new ConcreteBytesRefIndexFieldData.Builder())
                 .put(Tuple.tuple("double", "array"), new DoubleArrayIndexFieldData.Builder())
+                .put(Tuple.tuple("long", "array"), new LongArrayIndexFieldData.Builder())
                 .immutableMap();
     }
 

src/main/java/org/elasticsearch/index/fielddata/plain/LongArrayAtomicFieldData.java

@@ -0,0 +1,913 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch 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.elasticsearch.index.fielddata.plain;
+
+import org.apache.lucene.util.FixedBitSet;
+import org.elasticsearch.common.RamUsage;
+import org.elasticsearch.index.fielddata.*;
+import org.elasticsearch.index.fielddata.ordinals.Ordinals;
+import org.elasticsearch.index.fielddata.util.DoubleArrayRef;
+import org.elasticsearch.index.fielddata.util.IntArrayRef;
+import org.elasticsearch.index.fielddata.util.LongArrayRef;
+import org.elasticsearch.index.fielddata.util.StringArrayRef;
+
+/**
+ */
+public abstract class LongArrayAtomicFieldData implements AtomicNumericFieldData {
+
+    protected final long[] values;
+    private final int numDocs;
+
+    protected long size = -1;
+
+    public LongArrayAtomicFieldData(long[] values, int numDocs) {
+        this.values = values;
+        this.numDocs = numDocs;
+    }
+
+    @Override
+    public int getNumDocs() {
+        return numDocs;
+    }
+
+    public static class WithOrdinals extends LongArrayAtomicFieldData {
+
+        private final Ordinals ordinals;
+
+        public WithOrdinals(long[] values, int numDocs, Ordinals ordinals) {
+            super(values, numDocs);
+            this.ordinals = ordinals;
+        }
+
+        @Override
+        public boolean isMultiValued() {
+            return ordinals.isMultiValued();
+        }
+
+        @Override
+        public boolean isValuesOrdered() {
+            return true;
+        }
+
+        @Override
+        public long getMemorySizeInBytes() {
+            if (size == -1) {
+                size = RamUsage.NUM_BYTES_INT/*size*/ + RamUsage.NUM_BYTES_INT/*numDocs*/ + +RamUsage.NUM_BYTES_ARRAY_HEADER + (values.length * RamUsage.NUM_BYTES_DOUBLE) + ordinals.getMemorySizeInBytes();
+            }
+            return size;
+        }
+
+        @Override
+        public BytesValues getBytesValues() {
+            return new BytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public HashedBytesValues getHashedBytesValues() {
+            return new HashedBytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public StringValues getStringValues() {
+            return new StringValues(values, ordinals.ordinals());
+        }
+
+        @Override
+        public ScriptDocValues getScriptValues() {
+            return new ScriptDocValues.NumericDouble(getDoubleValues());
+        }
+
+        @Override
+        public ByteValues getByteValues() {
+            return new ByteValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public ShortValues getShortValues() {
+            return new ShortValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public IntValues getIntValues() {
+            return new IntValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public LongValues getLongValues() {
+            return new LongValues(values, ordinals.ordinals());
+        }
+
+        @Override
+        public FloatValues getFloatValues() {
+            return new FloatValues.DoubleBased(getDoubleValues());
+        }
+
+        @Override
+        public DoubleValues getDoubleValues() {
+            return new DoubleValues(values, ordinals.ordinals());
+        }
+
+        static class StringValues implements org.elasticsearch.index.fielddata.StringValues {
+
+            private final long[] values;
+            private final Ordinals.Docs ordinals;
+
+            private final StringArrayRef arrayScratch = new StringArrayRef(new String[1], 1);
+            private final ValuesIter valuesIter;
+
+            StringValues(long[] values, Ordinals.Docs ordinals) {
+                this.values = values;
+                this.ordinals = ordinals;
+                this.valuesIter = new ValuesIter(values);
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return ordinals.getOrd(docId) != 0;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return ordinals.isMultiValued();
+            }
+
+            @Override
+            public String getValue(int docId) {
+                int ord = ordinals.getOrd(docId);
+                if (ord == 0) {
+                    return null;
+                }
+                return Long.toString(values[ord]);
+            }
+
+            @Override
+            public StringArrayRef getValues(int docId) {
+                IntArrayRef ords = ordinals.getOrds(docId);
+                int size = ords.size();
+                if (size == 0) return StringArrayRef.EMPTY;
+
+                arrayScratch.reset(size);
+                for (int i = ords.start; i < ords.end; i++) {
+                    arrayScratch.values[arrayScratch.end++] = Long.toString(values[ords.values[i]]);
+                }
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return valuesIter.reset(ordinals.getIter(docId));
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                Ordinals.Docs.Iter iter = ordinals.getIter(docId);
+                int ord = iter.next();
+                if (ord == 0) {
+                    proc.onMissing(docId);
+                    return;
+                }
+                do {
+                    proc.onValue(docId, Long.toString(values[ord]));
+                } while ((ord = iter.next()) != 0);
+            }
+
+            static class ValuesIter implements Iter {
+
+                private final long[] values;
+                private Ordinals.Docs.Iter ordsIter;
+                private int ord;
+
+                ValuesIter(long[] values) {
+                    this.values = values;
+                }
+
+                public ValuesIter reset(Ordinals.Docs.Iter ordsIter) {
+                    this.ordsIter = ordsIter;
+                    this.ord = ordsIter.next();
+                    return this;
+                }
+
+                @Override
+                public boolean hasNext() {
+                    return ord != 0;
+                }
+
+                @Override
+                public String next() {
+                    String value = Long.toString(values[ord]);
+                    ord = ordsIter.next();
+                    return value;
+                }
+            }
+        }
+
+        static class LongValues implements org.elasticsearch.index.fielddata.LongValues {
+
+            private final long[] values;
+            private final Ordinals.Docs ordinals;
+
+            private final LongArrayRef arrayScratch = new LongArrayRef(new long[1], 1);
+            private final ValuesIter iter;
+
+            LongValues(long[] values, Ordinals.Docs ordinals) {
+                this.values = values;
+                this.ordinals = ordinals;
+                this.iter = new ValuesIter(values);
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return ordinals.isMultiValued();
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return ordinals.getOrd(docId) != 0;
+            }
+
+            @Override
+            public long getValue(int docId) {
+                return values[ordinals.getOrd(docId)];
+            }
+
+            @Override
+            public long getValueMissing(int docId, long missingValue) {
+                int ord = ordinals.getOrd(docId);
+                if (ord == 0) {
+                    return missingValue;
+                } else {
+                    return values[ord];
+                }
+            }
+
+            @Override
+            public LongArrayRef getValues(int docId) {
+                IntArrayRef ords = ordinals.getOrds(docId);
+                int size = ords.size();
+                if (size == 0) return LongArrayRef.EMPTY;
+
+                arrayScratch.reset(size);
+                for (int i = ords.start; i < ords.end; i++) {
+                    arrayScratch.values[arrayScratch.end++] = values[ords.values[i]];
+                }
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return iter.reset(ordinals.getIter(docId));
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                Ordinals.Docs.Iter iter = ordinals.getIter(docId);
+                int ord = iter.next();
+                if (ord == 0) {
+                    proc.onMissing(docId);
+                    return;
+                }
+                do {
+                    proc.onValue(docId, values[ord]);
+                } while ((ord = iter.next()) != 0);
+            }
+
+            static class ValuesIter implements Iter {
+
+                private final long[] values;
+                private Ordinals.Docs.Iter ordsIter;
+                private int ord;
+
+                ValuesIter(long[] values) {
+                    this.values = values;
+                }
+
+                public ValuesIter reset(Ordinals.Docs.Iter ordsIter) {
+                    this.ordsIter = ordsIter;
+                    this.ord = ordsIter.next();
+                    return this;
+                }
+
+                @Override
+                public boolean hasNext() {
+                    return ord != 0;
+                }
+
+                @Override
+                public long next() {
+                    long value = values[ord];
+                    ord = ordsIter.next();
+                    return value;
+                }
+            }
+        }
+
+        static class DoubleValues implements org.elasticsearch.index.fielddata.DoubleValues {
+
+            private final long[] values;
+            private final Ordinals.Docs ordinals;
+
+            private final DoubleArrayRef arrayScratch = new DoubleArrayRef(new double[1], 1);
+            private final ValuesIter iter;
+
+            DoubleValues(long[] values, Ordinals.Docs ordinals) {
+                this.values = values;
+                this.ordinals = ordinals;
+                this.iter = new ValuesIter(values);
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return ordinals.isMultiValued();
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return ordinals.getOrd(docId) != 0;
+            }
+
+            @Override
+            public double getValue(int docId) {
+                return (double) values[ordinals.getOrd(docId)];
+            }
+
+            @Override
+            public double getValueMissing(int docId, double missingValue) {
+                int ord = ordinals.getOrd(docId);
+                if (ord == 0) {
+                    return missingValue;
+                } else {
+                    return (double) values[ord];
+                }
+            }
+
+            @Override
+            public DoubleArrayRef getValues(int docId) {
+                IntArrayRef ords = ordinals.getOrds(docId);
+                int size = ords.size();
+                if (size == 0) return DoubleArrayRef.EMPTY;
+
+                arrayScratch.reset(size);
+                for (int i = ords.start; i < ords.end; i++) {
+                    arrayScratch.values[arrayScratch.end++] = (double) values[ords.values[i]];
+                }
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return iter.reset(ordinals.getIter(docId));
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                Ordinals.Docs.Iter iter = ordinals.getIter(docId);
+                int ord = iter.next();
+                if (ord == 0) {
+                    proc.onMissing(docId);
+                    return;
+                }
+                do {
+                    proc.onValue(docId, (double) values[ord]);
+                } while ((ord = iter.next()) != 0);
+            }
+
+            static class ValuesIter implements Iter {
+
+                private final long[] values;
+                private Ordinals.Docs.Iter ordsIter;
+                private int ord;
+
+                ValuesIter(long[] values) {
+                    this.values = values;
+                }
+
+                public ValuesIter reset(Ordinals.Docs.Iter ordsIter) {
+                    this.ordsIter = ordsIter;
+                    this.ord = ordsIter.next();
+                    return this;
+                }
+
+                @Override
+                public boolean hasNext() {
+                    return ord != 0;
+                }
+
+                @Override
+                public double next() {
+                    double value = (double) values[ord];
+                    ord = ordsIter.next();
+                    return value;
+                }
+            }
+        }
+    }
+
+    /**
+     * A single valued case, where not all values are "set", so we have a FixedBitSet that
+     * indicates which values have an actual value.
+     */
+    public static class SingleFixedSet extends LongArrayAtomicFieldData {
+
+        private final FixedBitSet set;
+
+        public SingleFixedSet(long[] values, int numDocs, FixedBitSet set) {
+            super(values, numDocs);
+            this.set = set;
+        }
+
+        @Override
+        public boolean isMultiValued() {
+            return false;
+        }
+
+        @Override
+        public boolean isValuesOrdered() {
+            return false;
+        }
+
+        @Override
+        public long getMemorySizeInBytes() {
+            if (size == -1) {
+                size = RamUsage.NUM_BYTES_ARRAY_HEADER + (values.length * RamUsage.NUM_BYTES_DOUBLE) + (set.getBits().length * RamUsage.NUM_BYTES_LONG);
+            }
+            return size;
+        }
+
+        @Override
+        public ScriptDocValues getScriptValues() {
+            return new ScriptDocValues.NumericDouble(getDoubleValues());
+        }
+
+        @Override
+        public BytesValues getBytesValues() {
+            return new BytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public HashedBytesValues getHashedBytesValues() {
+            return new HashedBytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public StringValues getStringValues() {
+            return new StringValues(values, set);
+        }
+
+        @Override
+        public ByteValues getByteValues() {
+            return new ByteValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public ShortValues getShortValues() {
+            return new ShortValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public IntValues getIntValues() {
+            return new IntValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public LongValues getLongValues() {
+            return new LongValues(values, set);
+        }
+
+        @Override
+        public FloatValues getFloatValues() {
+            return new FloatValues.DoubleBased(getDoubleValues());
+        }
+
+        @Override
+        public DoubleValues getDoubleValues() {
+            return new DoubleValues(values, set);
+        }
+
+        static class StringValues implements org.elasticsearch.index.fielddata.StringValues {
+
+            private final long[] values;
+            private final FixedBitSet set;
+
+            private final StringArrayRef arrayScratch = new StringArrayRef(new String[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            StringValues(long[] values, FixedBitSet set) {
+                this.values = values;
+                this.set = set;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return set.get(docId);
+            }
+
+            @Override
+            public String getValue(int docId) {
+                if (set.get(docId)) {
+                    return Long.toString(values[docId]);
+                } else {
+                    return null;
+                }
+            }
+
+            @Override
+            public StringArrayRef getValues(int docId) {
+                if (set.get(docId)) {
+                    arrayScratch.values[0] = Long.toString(values[docId]);
+                    return arrayScratch;
+                } else {
+                    return StringArrayRef.EMPTY;
+                }
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                if (set.get(docId)) {
+                    return iter.reset(Long.toString(values[docId]));
+                } else {
+                    return Iter.Empty.INSTANCE;
+                }
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                if (set.get(docId)) {
+                    proc.onValue(docId, Long.toString(values[docId]));
+                } else {
+                    proc.onMissing(docId);
+                }
+            }
+        }
+
+        static class LongValues implements org.elasticsearch.index.fielddata.LongValues {
+
+            private final long[] values;
+            private final FixedBitSet set;
+
+            private final LongArrayRef arrayScratch = new LongArrayRef(new long[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            LongValues(long[] values, FixedBitSet set) {
+                this.values = values;
+                this.set = set;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return set.get(docId);
+            }
+
+            @Override
+            public long getValue(int docId) {
+                return values[docId];
+            }
+
+            @Override
+            public long getValueMissing(int docId, long missingValue) {
+                if (set.get(docId)) {
+                    return values[docId];
+                } else {
+                    return missingValue;
+                }
+            }
+
+            @Override
+            public LongArrayRef getValues(int docId) {
+                if (set.get(docId)) {
+                    arrayScratch.values[0] = values[docId];
+                    return arrayScratch;
+                } else {
+                    return LongArrayRef.EMPTY;
+                }
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                if (set.get(docId)) {
+                    return iter.reset(values[docId]);
+                } else {
+                    return Iter.Empty.INSTANCE;
+                }
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                if (set.get(docId)) {
+                    proc.onValue(docId, values[docId]);
+                }
+            }
+        }
+
+        static class DoubleValues implements org.elasticsearch.index.fielddata.DoubleValues {
+
+            private final long[] values;
+            private final FixedBitSet set;
+
+            private final DoubleArrayRef arrayScratch = new DoubleArrayRef(new double[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            DoubleValues(long[] values, FixedBitSet set) {
+                this.values = values;
+                this.set = set;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return set.get(docId);
+            }
+
+            @Override
+            public double getValue(int docId) {
+                return (double) values[docId];
+            }
+
+            @Override
+            public double getValueMissing(int docId, double missingValue) {
+                if (set.get(docId)) {
+                    return (double) values[docId];
+                } else {
+                    return missingValue;
+                }
+            }
+
+            @Override
+            public DoubleArrayRef getValues(int docId) {
+                if (set.get(docId)) {
+                    arrayScratch.values[0] = (double) values[docId];
+                    return arrayScratch;
+                } else {
+                    return DoubleArrayRef.EMPTY;
+                }
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                if (set.get(docId)) {
+                    return iter.reset((double) values[docId]);
+                } else {
+                    return Iter.Empty.INSTANCE;
+                }
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                if (set.get(docId)) {
+                    proc.onValue(docId, (double) values[docId]);
+                }
+            }
+        }
+    }
+
+    /**
+     * Assumes all the values are "set", and docId is used as the index to the value array.
+     */
+    public static class Single extends LongArrayAtomicFieldData {
+
+        /**
+         * Note, here, we assume that there is no offset by 1 from docId, so position 0
+         * is the value for docId 0.
+         */
+        public Single(long[] values, int numDocs) {
+            super(values, numDocs);
+        }
+
+        @Override
+        public boolean isMultiValued() {
+            return false;
+        }
+
+        @Override
+        public boolean isValuesOrdered() {
+            return false;
+        }
+
+        @Override
+        public long getMemorySizeInBytes() {
+            if (size == -1) {
+                size = RamUsage.NUM_BYTES_ARRAY_HEADER + (values.length * RamUsage.NUM_BYTES_DOUBLE);
+            }
+            return size;
+        }
+
+        @Override
+        public ScriptDocValues getScriptValues() {
+            return new ScriptDocValues.NumericDouble(getDoubleValues());
+        }
+
+        @Override
+        public BytesValues getBytesValues() {
+            return new BytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public HashedBytesValues getHashedBytesValues() {
+            return new HashedBytesValues.StringBased(getStringValues());
+        }
+
+        @Override
+        public StringValues getStringValues() {
+            return new StringValues(values);
+        }
+
+        @Override
+        public ByteValues getByteValues() {
+            return new ByteValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public ShortValues getShortValues() {
+            return new ShortValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public IntValues getIntValues() {
+            return new IntValues.LongBased(getLongValues());
+        }
+
+        @Override
+        public LongValues getLongValues() {
+            return new LongValues(values);
+        }
+
+        @Override
+        public FloatValues getFloatValues() {
+            return new FloatValues.DoubleBased(getDoubleValues());
+        }
+
+        @Override
+        public DoubleValues getDoubleValues() {
+            return new DoubleValues(values);
+        }
+
+        static class StringValues implements org.elasticsearch.index.fielddata.StringValues {
+
+            private final long[] values;
+
+            private final StringArrayRef arrayScratch = new StringArrayRef(new String[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            StringValues(long[] values) {
+                this.values = values;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return true;
+            }
+
+            @Override
+            public String getValue(int docId) {
+                return Long.toString(values[docId]);
+            }
+
+            @Override
+            public StringArrayRef getValues(int docId) {
+                arrayScratch.values[0] = Long.toString(values[docId]);
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return iter.reset(Long.toString(values[docId]));
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                proc.onValue(docId, Long.toString(values[docId]));
+            }
+        }
+
+        static class LongValues implements org.elasticsearch.index.fielddata.LongValues {
+
+            private final long[] values;
+
+            private final LongArrayRef arrayScratch = new LongArrayRef(new long[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            LongValues(long[] values) {
+                this.values = values;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return true;
+            }
+
+            @Override
+            public long getValue(int docId) {
+                return values[docId];
+            }
+
+            @Override
+            public long getValueMissing(int docId, long missingValue) {
+                return values[docId];
+            }
+
+            @Override
+            public LongArrayRef getValues(int docId) {
+                arrayScratch.values[0] = values[docId];
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return iter.reset(values[docId]);
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                proc.onValue(docId, values[docId]);
+            }
+        }
+
+        static class DoubleValues implements org.elasticsearch.index.fielddata.DoubleValues {
+
+            private final long[] values;
+
+            private final DoubleArrayRef arrayScratch = new DoubleArrayRef(new double[1], 1);
+            private final Iter.Single iter = new Iter.Single();
+
+            DoubleValues(long[] values) {
+                this.values = values;
+            }
+
+            @Override
+            public boolean isMultiValued() {
+                return false;
+            }
+
+            @Override
+            public boolean hasValue(int docId) {
+                return true;
+            }
+
+            @Override
+            public double getValue(int docId) {
+                return (double) values[docId];
+            }
+
+            @Override
+            public double getValueMissing(int docId, double missingValue) {
+                return (double) values[docId];
+            }
+
+            @Override
+            public DoubleArrayRef getValues(int docId) {
+                arrayScratch.values[0] = (double) values[docId];
+                return arrayScratch;
+            }
+
+            @Override
+            public Iter getIter(int docId) {
+                return iter.reset((double) values[docId]);
+            }
+
+            @Override
+            public void forEachValueInDoc(int docId, ValueInDocProc proc) {
+                proc.onValue(docId, (double) values[docId]);
+            }
+        }
+    }
+}

src/main/java/org/elasticsearch/index/fielddata/plain/LongArrayIndexFieldData.java

@@ -0,0 +1,156 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch 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.elasticsearch.index.fielddata.plain;
+
+import gnu.trove.list.array.TLongArrayList;
+import org.apache.lucene.index.*;
+import org.apache.lucene.search.FieldCache;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.FixedBitSet;
+import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.fielddata.*;
+import org.elasticsearch.index.fielddata.fieldcomparator.DoubleValuesComparatorSource;
+import org.elasticsearch.index.fielddata.ordinals.MultiFlatArrayOrdinals;
+import org.elasticsearch.index.settings.IndexSettings;
+
+import java.util.ArrayList;
+
+/**
+ */
+public class LongArrayIndexFieldData extends AbstractIndexFieldData<LongArrayAtomicFieldData> implements IndexNumericFieldData<LongArrayAtomicFieldData> {
+
+    public static class Builder implements IndexFieldData.Builder {
+
+        @Override
+        public IndexFieldData build(Index index, @IndexSettings Settings indexSettings, String fieldName, FieldDataType type, IndexFieldDataCache cache) {
+            return new LongArrayIndexFieldData(index, indexSettings, fieldName, type, cache);
+        }
+    }
+
+    public LongArrayIndexFieldData(Index index, @IndexSettings Settings indexSettings, String fieldName, FieldDataType fieldDataType, IndexFieldDataCache cache) {
+        super(index, indexSettings, fieldName, fieldDataType, cache);
+    }
+
+    @Override
+    public NumericType getNumericType() {
+        return NumericType.DOUBLE;
+    }
+
+    @Override
+    public boolean valuesOrdered() {
+        // because we might have single values? we can dynamically update a flag to reflect that
+        // based on the atomic field data loaded
+        return false;
+    }
+
+    @Override
+    public LongArrayAtomicFieldData load(AtomicReaderContext context) {
+        try {
+            return cache.load(context, this);
+        } catch (Throwable e) {
+            if (e instanceof ElasticSearchException) {
+                throw (ElasticSearchException) e;
+            } else {
+                throw new ElasticSearchException(e.getMessage(), e);
+            }
+        }
+    }
+
+    @Override
+    public LongArrayAtomicFieldData loadDirect(AtomicReaderContext context) throws Exception {
+        AtomicReader reader = context.reader();
+
+        Terms terms = reader.terms(getFieldName());
+        if (terms == null) {
+            return new LongArrayAtomicFieldData.Single(new long[0], 0);
+        }
+
+        // TODO: how can we guess the number of terms? numerics end up creating more terms per value...
+        final TLongArrayList values = new TLongArrayList();
+        ArrayList<int[]> ordinals = new ArrayList<int[]>();
+        int[] idx = new int[reader.maxDoc()];
+        ordinals.add(new int[reader.maxDoc()]);
+
+        values.add(0); // first "t" indicates null value
+        int termOrd = 1;  // current term number
+
+        TermsEnum termsEnum = terms.iterator(null);
+        try {
+            DocsEnum docsEnum = null;
+            for (BytesRef term = termsEnum.next(); term != null; term = termsEnum.next()) {
+                values.add(FieldCache.NUMERIC_UTILS_LONG_PARSER.parseLong(term));
+                docsEnum = termsEnum.docs(reader.getLiveDocs(), docsEnum, 0);
+                for (int docId = docsEnum.nextDoc(); docId != DocsEnum.NO_MORE_DOCS; docId = docsEnum.nextDoc()) {
+                    int[] ordinal;
+                    if (idx[docId] >= ordinals.size()) {
+                        ordinal = new int[reader.maxDoc()];
+                        ordinals.add(ordinal);
+                    } else {
+                        ordinal = ordinals.get(idx[docId]);
+                    }
+                    ordinal[docId] = termOrd;
+                    idx[docId]++;
+                }
+                termOrd++;
+            }
+        } catch (RuntimeException e) {
+            if (e.getClass().getName().endsWith("StopFillCacheException")) {
+                // all is well, in case numeric parsers are used.
+            } else {
+                throw e;
+            }
+        }
+
+        if (ordinals.size() == 1) {
+            int[] nativeOrdinals = ordinals.get(0);
+            FixedBitSet set = new FixedBitSet(reader.maxDoc());
+            long[] sValues = new long[reader.maxDoc()];
+            boolean allHaveValue = true;
+            for (int i = 0; i < nativeOrdinals.length; i++) {
+                int nativeOrdinal = nativeOrdinals[i];
+                if (nativeOrdinal == 0) {
+                    allHaveValue = false;
+                } else {
+                    set.set(i);
+                    sValues[i] = values.get(nativeOrdinal);
+                }
+            }
+            if (allHaveValue) {
+                return new LongArrayAtomicFieldData.Single(sValues, reader.maxDoc());
+            } else {
+                return new LongArrayAtomicFieldData.SingleFixedSet(sValues, reader.maxDoc(), set);
+            }
+        } else {
+            int[][] nativeOrdinals = new int[ordinals.size()][];
+            for (int i = 0; i < nativeOrdinals.length; i++) {
+                nativeOrdinals[i] = ordinals.get(i);
+            }
+            return new LongArrayAtomicFieldData.WithOrdinals(values.toArray(new long[values.size()]), reader.maxDoc(), new MultiFlatArrayOrdinals(nativeOrdinals, termOrd));
+        }
+    }
+
+    @Override
+    public XFieldComparatorSource comparatorSource(@Nullable Object missingValue) {
+        return new DoubleValuesComparatorSource(this, missingValue);
+    }
+}

src/test/java/org/elasticsearch/test/unit/index/fielddata/DoubleFieldDataTests.java

@@ -19,6 +19,10 @@
 
 package org.elasticsearch.test.unit.index.fielddata;
 
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.DoubleField;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
 import org.elasticsearch.index.fielddata.FieldDataType;
 
 /**
@@ -29,4 +33,95 @@ public class DoubleFieldDataTests extends NumericFieldDataTests {
     protected FieldDataType getFieldDataType() {
         return new FieldDataType("double");
     }
+
+    protected String one() {
+        return "1.0";
+    }
+
+    protected String two() {
+        return "2.0";
+    }
+
+    protected String three() {
+        return "3.0";
+    }
+
+    protected String four() {
+        return "4.0";
+    }
+
+
+    @Override
+    protected void fillSingleValueAllSet() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        d.add(new DoubleField("value", 1.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillSingleValueWithMissing() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING....
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillMultiValueAllSet() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
+        d.add(new DoubleField("value", 4.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        d.add(new DoubleField("value", 1.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillMultiValueWithMissing() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
+        d.add(new DoubleField("value", 4.0d, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new DoubleField("value", 3.0f, Field.Store.NO));
+        writer.addDocument(d);
+    }
 }

src/test/java/org/elasticsearch/test/unit/index/fielddata/LongFieldDataTests.java

@@ -0,0 +1,110 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch 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.elasticsearch.test.unit.index.fielddata;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.LongField;
+import org.apache.lucene.document.StringField;
+import org.elasticsearch.index.fielddata.FieldDataType;
+
+/**
+ */
+public class LongFieldDataTests extends NumericFieldDataTests {
+
+    @Override
+    protected FieldDataType getFieldDataType() {
+        return new FieldDataType("long");
+    }
+
+    @Override
+    protected void fillSingleValueAllSet() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new LongField("value", 2, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        d.add(new LongField("value", 1, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new LongField("value", 3, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillSingleValueWithMissing() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new LongField("value", 2, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING....
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new LongField("value", 3, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillMultiValueAllSet() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new LongField("value", 2, Field.Store.NO));
+        d.add(new LongField("value", 4, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        d.add(new LongField("value", 1, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new LongField("value", 3, Field.Store.NO));
+        writer.addDocument(d);
+    }
+
+    @Override
+    protected void fillMultiValueWithMissing() throws Exception {
+        Document d = new Document();
+        d.add(new StringField("_id", "1", Field.Store.NO));
+        d.add(new LongField("value", 2, Field.Store.NO));
+        d.add(new LongField("value", 4, Field.Store.NO));
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "2", Field.Store.NO));
+        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING
+        writer.addDocument(d);
+
+        d = new Document();
+        d.add(new StringField("_id", "3", Field.Store.NO));
+        d.add(new LongField("value", 3, Field.Store.NO));
+        writer.addDocument(d);
+    }
+}

src/test/java/org/elasticsearch/test/unit/index/fielddata/NumericFieldDataTests.java

@@ -19,10 +19,6 @@
 
 package org.elasticsearch.test.unit.index.fielddata;
 
-import org.apache.lucene.document.Document;
-import org.apache.lucene.document.DoubleField;
-import org.apache.lucene.document.Field;
-import org.apache.lucene.document.StringField;
 import org.apache.lucene.search.*;
 import org.elasticsearch.index.fielddata.*;
 import org.elasticsearch.index.fielddata.util.*;
@@ -38,40 +34,6 @@ public abstract class NumericFieldDataTests extends StringFieldDataTests {
 
     protected abstract FieldDataType getFieldDataType();
 
-    protected String one() {
-        return "1.0";
-    }
-
-    protected String two() {
-        return "2.0";
-    }
-
-    protected String three() {
-        return "3.0";
-    }
-
-    protected String four() {
-        return "4.0";
-    }
-
-    @Override
-    protected void fillSingleValueAllSet() throws Exception {
-        Document d = new Document();
-        d.add(new StringField("_id", "1", Field.Store.NO));
-        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "2", Field.Store.NO));
-        d.add(new DoubleField("value", 1.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "3", Field.Store.NO));
-        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
-        writer.addDocument(d);
-    }
-
     @Test
     public void testSingleValueAllSetNumber() throws Exception {
         fillSingleValueAllSet();
@@ -380,24 +342,6 @@ public abstract class NumericFieldDataTests extends StringFieldDataTests {
         assertThat(topDocs.scoreDocs[2].doc, equalTo(1));
     }
 
-    @Override
-    protected void fillSingleValueWithMissing() throws Exception {
-        Document d = new Document();
-        d.add(new StringField("_id", "1", Field.Store.NO));
-        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "2", Field.Store.NO));
-        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING....
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "3", Field.Store.NO));
-        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
-        writer.addDocument(d);
-    }
-
     @Test
     public void testSingleValueWithMissingNumber() throws Exception {
         fillSingleValueWithMissing();
@@ -710,25 +654,6 @@ public abstract class NumericFieldDataTests extends StringFieldDataTests {
         assertThat(topDocs.scoreDocs[2].doc, equalTo(1));
     }
 
-    @Override
-    protected void fillMultiValueAllSet() throws Exception {
-        Document d = new Document();
-        d.add(new StringField("_id", "1", Field.Store.NO));
-        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
-        d.add(new DoubleField("value", 4.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "2", Field.Store.NO));
-        d.add(new DoubleField("value", 1.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "3", Field.Store.NO));
-        d.add(new DoubleField("value", 3.0d, Field.Store.NO));
-        writer.addDocument(d);
-    }
-
     @Test
     public void testMultiValueAllSetNumber() throws Exception {
         fillMultiValueAllSet();
@@ -1038,25 +963,6 @@ public abstract class NumericFieldDataTests extends StringFieldDataTests {
         floatValues.forEachValueInDoc(2, new FloatValuesVerifierProc(2).addExpected(3f));
     }
 
-    @Override
-    protected void fillMultiValueWithMissing() throws Exception {
-        Document d = new Document();
-        d.add(new StringField("_id", "1", Field.Store.NO));
-        d.add(new DoubleField("value", 2.0d, Field.Store.NO));
-        d.add(new DoubleField("value", 4.0d, Field.Store.NO));
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "2", Field.Store.NO));
-        //d.add(new StringField("value", one(), Field.Store.NO)); // MISSING
-        writer.addDocument(d);
-
-        d = new Document();
-        d.add(new StringField("_id", "3", Field.Store.NO));
-        d.add(new DoubleField("value", 3.0f, Field.Store.NO));
-        writer.addDocument(d);
-    }
-
     @Test
     public void testMultiValueWithMissingNumber() throws Exception {
         fillMultiValueWithMissing();