@Override
    protected void extractAndTakeAction(String timestamp, JSONArray jsonAllRows, List<String> joinFields, RequestType requestType, ActionType action) {
        Map<Object,List<Object>> newBaseAllRows = new LinkedHashMap<>();
        JSONObject eachRow = null;
        for (int i = 0; i < jsonAllRows.length(); i++) {
            JSONObject jsonItem = jsonAllRows.getJSONObject(i);
            eachRow = dataItemAt(jsonItem, requestType, action);
            String actualTimestamp = timestamp;
            if (timestamp == null) {
                if (requestType == RequestType.SELECT) {
                    actualTimestamp = eachRow.optString("timestamp");
                    // Because the timestamp is within each row remove them once you extract it.
                    eachRow.remove("timestamp");
                } else {
                    actualTimestamp = jsonItem.optString("timestamp");
                }
            }
            Tuple2<Object, List<Object>> row = mapPkToRow(actualTimestamp, eachRow, joinFields);
            Object pk = row._1();// Primary key.
            List<Object> rowVal = row._2();
            if (action == ActionType.FIRST_CUT) {// First cut (no join).
                if (pk != null) {//TODO: Only non-nulls will go through. Address this later.
                    baseAllRows.put(pk, rowVal);
                }
            } else {// Some Join action.
                if (pk != null) {//TODO: Only non-nulls will go through. Address this later.
                    if (baseAllRows.containsKey(pk)) {// If any item in set B exist in set A.
                        rowVal.removeAll(splitCompositeKeys(pk.toString()));// Remove key(which is already available in baseAllRows)
                        if (!joinFields.contains("timestamp")) {// If join field is not "timestamp" then remove timestamp data value from to be joined set.
                            rowVal.remove(0);// Because the 1st field is always timestamp.(See extractKeyAndRow() method)
                        }
                        if (action == ActionType.JOIN || action == ActionType.RIGHT_JOIN) {
                            newBaseAllRows.put(pk, baseAllRows.remove(pk));// Remove from existing map and add to new map.
                            newBaseAllRows.get(pk).addAll(rowVal);// Update the new map with partial to be joined data.
                        } else if (action == ActionType.LEFT_JOIN) {// Left join
                            baseAllRows.get(pk).addAll(rowVal);
                        }
                    } else {// For right join we still need to continue with each row in set B.
                        if (action == ActionType.RIGHT_JOIN) {// Right join
                            newBaseAllRows.put(pk, rowVal);
                        }
                    }
                }
            }
            if (i == 0) {// Fill headers (only once)
                fillHeaders(eachRow, joinFields, action);
            }
        }
        if (!newBaseAllRows.isEmpty()) {// For join and right join.
            baseAllRows = newBaseAllRows;
        }
    }
@ Override protected void extractAndTakeAction ( String timestamp , JSONArray jsonAllRows , List < String > joinFields , RequestType requestType , ActionType action ) { Map < Object , List < Object > > newBaseAllRows = new LinkedHashMap < > ( ) ; JSONObject eachRow = null ; for ( int i = 0 ; i < jsonAllRows . length ( ) ; i ++ ) { JSONObject jsonItem = jsonAllRows . getJSONObject ( i ) ; eachRow = dataItemAt ( jsonItem , requestType , action ) ; String actualTimestamp = timestamp ; if ( timestamp == null ) { if ( requestType == RequestType . SELECT ) { actualTimestamp = eachRow . optString ( "timestamp" ) ; eachRow . remove ( "timestamp" ) ; } else { actualTimestamp = jsonItem . optString ( "timestamp" ) ; } } Tuple2 < Object , List < Object > > row = mapPkToRow ( actualTimestamp , eachRow , joinFields ) ; Object pk = row . _1 ( ) ; List < Object > rowVal = row . _2 ( ) ; if ( action == ActionType . FIRST_CUT ) { if ( pk != null ) { baseAllRows . put ( pk , rowVal ) ; } } else { if ( pk != null ) { if ( baseAllRows . containsKey ( pk ) ) { rowVal . removeAll ( splitCompositeKeys ( pk . toString ( ) ) ) ; if ( ! joinFields . contains ( "timestamp" ) ) { rowVal . remove ( 0 ) ; } if ( action == ActionType . JOIN || action == ActionType . RIGHT_JOIN ) { newBaseAllRows . put ( pk , baseAllRows . remove ( pk ) ) ; newBaseAllRows . get ( pk ) . addAll ( rowVal ) ; } else if ( action == ActionType . LEFT_JOIN ) { baseAllRows . get ( pk ) . addAll ( rowVal ) ; } } else { if ( action == ActionType . RIGHT_JOIN ) { newBaseAllRows . put ( pk , rowVal ) ; } } } } if ( i == 0 ) { fillHeaders ( eachRow , joinFields , action ) ; } } if ( ! newBaseAllRows . isEmpty ( ) ) { baseAllRows = newBaseAllRows ; } }
Generate a Tuple <A, B>
A = list of field names
B = map of <joinFeild, rowList> from jsonArray.
@param timestamp
@param jsonAllRows
@param joinFields
@param requestType
@param action