package io.github.adrianulbona.osm.parquet.convertor;

import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
import org.openstreetmap.osmosis.core.domain.v0_6.Way;

import java.util.ArrayList;
import java.util.List;

import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
import static org.apache.parquet.schema.Type.Repetition.REPEATED;


/**
 * Created by adrian.bona on 26/03/16.
 */
public class WayWriteSupport extends OsmEntityWriteSupport<Way> {

    private final PrimitiveType nodesType;

    public WayWriteSupport() {
        nodesType = new PrimitiveType(REPEATED, INT64, "nodes");
    }

    @Override
    protected MessageType getSchema() {
        final List<Type> attributes = new ArrayList<>(getCommonAttributes());
        attributes.add(nodesType);
        return new MessageType("way", attributes);
    }

    @Override
    protected void writeSpecificFields(Way record, int nextAvailableIndex) {
        recordConsumer.startField(nodesType.getName(), nextAvailableIndex);
        record.getWayNodes().forEach(wayNode -> recordConsumer.addLong(wayNode.getNodeId()));
        recordConsumer.endField(nodesType.getName(), nextAvailableIndex);
    }
}

@@ -48,16 +48,60 @@
 ertor;%0A%0A
+import org.apache.parquet.schema.GroupType;%0A
 import o
@@ -284,60 +284,294 @@
 ay;%0A
-%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0A
+import org.openstreetmap.osmosis.core.domain.v0_6.WayNode;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0A%0Aimport static java.util.stream.IntStream.range;%0Aimport static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
 %0Aimp
@@ -712,16 +712,82 @@
 PEATED;%0A
+import static org.apache.parquet.schema.Type.Repetition.REQUIRED;%0A
 %0A%0A/**%0A *
@@ -934,14 +934,97 @@
 node
-s
+Index
 Type;
+%0A    private final PrimitiveType nodeIdType;%0A    private final GroupType nodes;
 %0A%0A  
@@ -1060,25 +1060,29 @@
         node
-s
+Index
 Type = new P
@@ -1100,30 +1100,174 @@
 e(RE
-PEATED, INT64, %22nodes%22
+QUIRED, INT32, %22index%22);%0A        nodeIdType = new PrimitiveType(REQUIRED, INT64, %22nodeId%22);%0A        nodes = new GroupType(REPEATED, %22nodes%22, nodeIndexType, nodeIdType
 );%0A 
@@ -1432,20 +1432,16 @@
 dd(nodes
-Type
 );%0A     
@@ -1580,24 +1580,304 @@
 bleIndex) %7B%0A
+        final List%3CWayNode%3E wayNodes = record.getWayNodes();%0A        final Map%3CInteger, Long%3E indexedNodes = new HashMap%3C%3E();%0A        range(0, wayNodes.size()).forEach(index -%3E indexedNodes.put(index, wayNodes.get(index).getNodeId()));%0A%0A        if (!indexedNodes.isEmpty()) %7B%0A    
         reco
@@ -1895,36 +1895,32 @@
 startField(nodes
-Type
 .getName(), next
@@ -1944,36 +1944,32 @@
         
-record.getWay
+    indexed
 Nodes
-()
 .forEach
@@ -1973,73 +1973,395 @@
 ach(
-wayNode -%3E recordConsumer.addLong(wayNode.getNodeId()));%0A
+(index, nodeId) -%3E %7B%0A                recordConsumer.startGroup();%0A%0A                recordConsumer.startField(nodeIndexType.getName(), 0);%0A                recordConsumer.addInteger(index);%0A                recordConsumer.endField(nodeIndexType.getName(), 0);%0A%0A                recordConsumer.startField(nodeIdType.getName(), 1);%0A                recordConsumer.addLong(nodeId);%0A     
         
 reco
@@ -2348,32 +2348,35 @@
 );%0A             
+   
 recordConsumer.e
@@ -2387,17 +2387,18 @@
 eld(node
-s
+Id
 Type.get
@@ -2396,32 +2396,149 @@
 dType.getName(),
+ 1);%0A%0A                recordConsumer.endGroup();%0A            %7D);%0A            recordConsumer.endField(nodes.getName(),
  nextAvailableIn
@@ -2539,20 +2539,30 @@
 ableIndex);%0A
+        %7D%0A
     %7D%0A%7D%0A

Add indices on way nodes