main.java

/**
 * Code generation of SQL query ====================================================================
 *
 * This exercise is a simplified example of the type of logic one needs to write when targeting
 * XONAI underlying compiler to accelerate big data computations. Data is presented in columnar
 * format (as opposed to default Spark row format) similarly to the format used by our engine. The
 * goal is to write the missing code that performs the computation of the query.
 *
 * ATTENTION:
 * It should be written with performance in mind where every extra instruction matters. Also use of
 * libraries or JDK classes/methods are discouraged as the role entails writing low-level libraries.
 * Feel free to create new functions but KEEP existing ones.
 *
 * Consider the following SQL query:
 *
 *   select
 *     100 * (sum(price * discount) / sum(price)) as discount_ratio,
 *   	 (avg(price) filter(where discount = 0.05)) as avg_price
 *   from
 *   	 item
 *   where (
 *   	 discount between .05 and .07
 *   	 and quantity < 24
 *   	 and status = `A`
 *   ) or comment LIKE `PROMO%SUMMER`
 *
 * It is composed by 3 main tasks:
 *   1. Read input
 *   2. Filter by WHERE condition
 *   3. Aggregate
 *
 * Traditionally data is represented in row format:
 *
 *   --------------------------------------------------------
 *    RowId   Quantity   Price   Discount   Status   Comment
 *   --------------------------------------------------------
 *    0       6          19.9    0.07       "A"      "PROMO"
 *   --------------------------------------------------------
 *    1       18         24.9    0.04       "A"      ""
 *   --------------------------------------------------------
 *    2       6          9.9     0.08       "AR"     ""
 *   --------------------------------------------------------
 *
 * This program represents data in columnar format:
 *
 *   -------------------------------------
 *    RowId      0         1         2
 *   -------------------------------------
 *    Quantity   6         18        6
 *   -------------------------------------
 *    Price      19.9      24.9      9.9
 *   -------------------------------------
 *    Discount   0.07      0.04      0.08
 *   -------------------------------------
 *    Status     "A"       "A"       "AR"
 *   -------------------------------------
 *    Comment    "PROMO"   ""        ""
 *   -------------------------------------
 */
public class Main {

  public static class InputBatch {
    public int numRows;
    public int[] quantity;
    public double[] price;
    public double[] discount;
    public StringColumn status = new StringColumn();
    public StringColumn comment = new StringColumn();
  }

  public static class FilteredBatch {
    public int numRows;
    public int[] quantity;
    public double[] price;
    public double[] discount;
    public StringColumn status = new StringColumn();
    public StringColumn comment = new StringColumn();
  }

  public static class AggregatedBatch {
    public int numRows;
    public double[] discount_ratio;
    public double[] avg_price;
  }

  /**
   * Variable length ASCII string.
   */
  public static class StringColumn {    
    public int[] offset; // start of string in `buffer` (for each row)
    public int[] length; // lengths (for each row)
    public byte[] buffer; // buffer with data of all strings

    // Example: 2 strings - "Hi" and "there"
    //
    // offset  |  0  2
    // length  |  2  5
    // buffer  |  H  i  t  h  e  r  e
  }

  private static String indexedString(StringColumn strColumn, int index) {
      int offset = strColumn.offset[index];
      int length = strColumn.length[index];
      return new String(strColumn.buffer, offset, length);
  }


  public static void main(String[] args) {
    testIsEqualToA();
    testIsLikePromoSummer();
    testQuery();
  }

  public static void testQuery() {
    InputBatch input = new InputBatch();

    input.numRows = 10;
    input.quantity = new int[]{ 6, 18, 6, 30, 24, 12, 18, 6, 24, 12 };
    input.price = new double[]{ 19.9d, 24.9d, 9.9d, 14.9d, 9.9d, 19.9d, 24.9d, 19.9d, 9.9d, 14.9 };
    input.discount = inputDiscount();
    
    input.status.offset = new int[]{ 0, 1, 2, 4, 5, 6, 7, 8, 9, 10 };
    input.status.length = new int[]{ 1, 1, 2, 1, 1, 1, 1, 1, 1, 1 };
    input.status.buffer = "AAARNAANAAA".getBytes();

    input.comment.offset = new int[]{ 0, 0, 0, 5, 0, 0, 0, 0, 20, 0 };
    input.comment.length = new int[]{ 5, 0, 0, 15, 5, 0, 0, 0, 12, 5 };
    input.comment.buffer = "PROMOPROMO IN SUMMERPROMO WINTER".getBytes();

    print(input);
    FilteredBatch filtered     = filter(input);
    print(filtered);
    AggregatedBatch aggregated = aggregate(filtered);
    print(aggregated);

    assertResult(5.8893854d, aggregated.discount_ratio[0], "discount ratio");
    assertResult(16.566666d, aggregated.avg_price[0], "avg price");
  }

  public static double[] inputDiscount() {
    // Possible discount values.
    double[] dictionary = { 0.04d, 0.05d, 0.07d, 0.08d };

    // Contains for each row the index in the discount dictionary.
    int[] ids = { 2, 0, 3, 1, 0, 2, 2, 1, 3, 1 };
    double[] expected = new double[ids.length];
    
    int index = 0;
    for (int idValue : ids) {
        expected[index++] = dictionary[idValue];
    }

    return expected;
  }

  public static byte[] stringByteArrayA = "A".getBytes();

  public static boolean isEqualToA(int rowId, StringColumn str) {
    return isEqualToByteArray(rowId, str, stringByteArrayA);
  }

  public static void testIsEqualToA() {
    StringColumn str = new StringColumn();

    str.offset = new int[]{ 0, 1, 2, 4, 5, 6, 7, 8, 9, 10 };
    str.length = new int[]{ 1, 1, 2, 1, 1, 1, 1, 1, 1, 1 };
    str.buffer = "AAARNAANAAA".getBytes();

    checkTest(
      isEqualToA(0, str),
      "0th row is A"
    );

    checkTest(
      isEqualToA(1, str),
      "1th row is A"
    );

    checkTest(
      ! isEqualToA(2, str),
      "2th row is not A"
    );

    checkTest(
      ! isEqualToA(6, str),
      "6th row is not A"
    );
  }

  public static boolean isLikePromoSummer(int rowId, StringColumn str) {
    return isLikeByteArray(rowId, str, "PROMO%SUMMER".getBytes());
  }

  public static void testIsLikePromoSummer() {
    StringColumn str = new StringColumn();

    str.offset = new int[]{ 0, 0, 0, 5, 0, 0, 0, 0, 20, 0 };
    str.length = new int[]{ 5, 0, 0, 15, 5, 0, 0, 0, 12, 5 };

    str.buffer = "PROMOPROMO IN SUMMERPROMO WINTER".getBytes();

    checkTest(
      isLikeByteArray(0, str, "PROMO".getBytes()),
      "0th, 'PROMO': LIKE 'PROMO'"
    );

    checkTest(
      !isLikeByteArray(0, str, "PROMO%SUMMER".getBytes()),
      "0th, 'PROMO': NOT LIKE 'PROMO%SUMMER'"
    );

    checkTest(
      !isLikeByteArray(3, str, "PROMO%FOO".getBytes()),
      "3th, 'PROMO IN SUMMER': LIKE 'PROMO%FOO'"
    );

    checkTest(
      isLikeByteArray(3, str, "PROMO%SUMMER".getBytes()),
      "3th, 'PROMO IN SUMMER': LIKE 'PROMO%SUMMER'"
    );

    checkTest(
      isLikeByteArray(3, str, "PROMO%".getBytes()),
      "3th, 'PROMO IN SUMMER': LIKE 'PROMO%'"
    );

    checkTest(
      !isLikeByteArray(3, str, "%PROMO".getBytes()),
      "3th, 'PROMO IN SUMMER': NOT LIKE '%PROMO'"
    );

    checkTest(
      isLikeByteArray(3, str, "%SUMMER".getBytes()),
      "3th, 'PROMO IN SUMMER': LIKE '%SUMMER'"
    );

    checkTest(
      isLikeByteArray(3, str, "PRO%IN%ER".getBytes()),
      "3th, 'PROMO IN SUMMER': LIKE 'PRO%IN%ER'"
    );

    checkTest(
      !isLikeByteArray(3, str, "%FOO".getBytes()),
      "3th, 'PROMO IN SUMMER': NOT LIKE '%FOO'"
    );

    checkTest(
      !isLikeByteArray(3, str, "PROMO%I%FOO".getBytes()),
      "3th, 'PROMO IN SUMMER': NOT LIKE 'PROMO%I%FOO'"
    );
  }

  public static boolean checkFirstFilter(int rowId, InputBatch input) {
    if (!isEqualToA(rowId, input.status))
        return false;
   
    if (24 <= input.quantity[rowId])
        return false;

    return isValueBetween(input.discount[rowId], 0.05, 0.07);
}


  /**
   * Applies filter: (
   *   discount between .05 and .07
   *   and quantity < 24
   *   and status = `A`
   * ) or comment LIKE `PROMO%SUMMER`
   */
  public static FilteredBatch filter(InputBatch input) {
    FilteredBatch output = new FilteredBatch();
    /*
     * Dangerous optimization:
     * To NOT allocate new space for FilteredBatch, It's possible to use input's space.
     * But the cost, credibility of InputBatch! We are overwriting its data and not respecting its ownership.
     * Can I call this C++'s equavalent's std::move?!
     * One catch: The filtering should be sequencial.
     */
    output.discount       = input.discount;         //new double[input.numRows];
    output.price          = input.price;            //new double[input.numRows];
    output.quantity       = input.quantity;         //new int[input.numRows];
    output.status.length  = input.status.length;    //new int[input.numRows];
    output.status.offset  = input.status.offset;    //new int[input.numRows];
    output.comment.length = input.comment.length;   //new int[input.numRows];
    output.comment.offset = input.comment.offset;   //new int[input.numRows];

    byte[] likePattern = "PROMO%SUMMER".getBytes();

    // Strings can be optimized but I am going to skip this part,
    // since it can be premature for this test.
    String commentBuffer = "";
    String statusBuffer = ""; 

    int outputRowId = 0;
    for (int rowId = 0; rowId < input.numRows; ++rowId) {

        // in X || Y, Y will not be evaluated if X is true
        boolean predicate = checkFirstFilter(rowId, input) || isLikeByteArray(rowId, input.comment, likePattern);
        if ( !predicate ) {
            continue;
        }

        //While filtering, functions such as "isLikeByteArray" or "checkFirstFilter" can 
        //carry over the value indexes for furthur optimization.
        output.price[outputRowId]     = input.price[rowId];
        output.discount[outputRowId]  = input.discount[rowId];
        output.quantity[outputRowId]  = input.quantity[rowId];

        int commentLength                  = input.comment.length[rowId];
        int commentOffset                  = input.comment.offset[rowId];
        output.comment.length[outputRowId] = commentLength;
        output.comment.offset[outputRowId] = commentBuffer.length();
        commentBuffer += new String(input.comment.buffer, commentOffset, commentLength); // text concatenation is error-prune, cheating with String

        int statusLength                   = input.status.length[rowId];
        int statusOffset                   = input.status.offset[rowId];
        output.status.length[outputRowId]  = statusLength;
        output.status.offset[outputRowId]  = statusBuffer.length();

        statusBuffer += new String(input.status.buffer, statusOffset, statusLength); // text concatenation is error-prune, cheating with String
        ++outputRowId;
    }

    output.comment.buffer = commentBuffer.getBytes(); // a little bit of cheating here.
    output.status.buffer  = statusBuffer.getBytes();
    output.numRows        = outputRowId; // DO NOT FORGET
    return output;
  }

  public static void print(InputBatch batch) {
      System.out.printf("%nINPUT:%n%10s %10s %10s %10s %10s%n", "Quantity", "Price", "Discount", "Status", "Comment");

      for (int index = 0; index < batch.numRows; index++) {
          int quantity = batch.quantity[index];
          double price    = batch.price[index];
          double discount = batch.discount[index];
          String status   = indexedString(batch.status, index); 
          String comment  = indexedString(batch.comment, index);

          System.out.printf("%10d %10f %10f %10s %10s %n", quantity, price, discount, status, comment);
      }
  }


  public static void print(FilteredBatch batch) {
      System.out.printf("%nFiltered:%n%10s %10s %10s %10s %10s%n", "Quantity", "Price", "Discount", "Status", "Comment");

      for (int index = 0; index < batch.numRows; index++) {
          int quantity = batch.quantity[index];
          double price    = batch.price[index];
          double discount = batch.discount[index];
          String status   = indexedString(batch.status, index); 
          String comment  = indexedString(batch.comment, index); 

          System.out.printf("%10d %10f %10f %10s %10s %n", quantity, price, discount, status, comment);
      }
  }


    public static void print(AggregatedBatch batch) {
      System.out.printf("%nAggregated:%n%20s %20s %n", "Price", "Discount");

      for (int index = 0; index < batch.numRows; index++) {
          double discount = batch.discount_ratio[index];
          double price    = batch.avg_price[index];

          System.out.printf("%20f %20f %n", price, discount);
      }
  }

  public static boolean isValueBetween(double value, double from, double to) {
      return (from <= value && value <= to);
  }


  public static boolean isEqualToByteArray(int rowId, StringColumn strColumn, byte[] buffer) {
      int length = strColumn.length[rowId];
      if (buffer.length != length) {
          return false;
      }

      int offset = strColumn.offset[rowId];
      int offsetLength = offset + length;
      int index = 0;

      while (index < length && (buffer[index] == strColumn.buffer[offset])) {
          ++offset;
          ++index;
      }

      return index == length && offsetLength == offset;
  }

  public static boolean isLikeByteArray(int rowId, StringColumn strColumn, byte[] pattern) {
      int length = strColumn.length[rowId];

      int offsetIndex = strColumn.offset[rowId];
      int offsetEnd = offsetIndex + length;
      int patternIndex = 0;
      int patternLength = pattern.length;

      boolean hardMatch = true;
      int lastMatch = 0;

      while (patternIndex < patternLength && offsetIndex < offsetEnd) {
          byte patternChar = pattern[patternIndex];

          if (patternChar == '%') {
              hardMatch  = false;
              lastMatch = ++patternIndex;

              if (lastMatch == patternLength)
                  offsetIndex = offsetEnd;

              continue;
          }

          byte textChar = strColumn.buffer[offsetIndex];
          boolean isMatched = textChar == patternChar;

          if (isMatched)
              ++patternIndex;
          else {
              patternIndex = lastMatch;
              if (hardMatch)
                  return false;
          }

          ++offsetIndex;
          if (patternIndex == patternLength && offsetIndex < offsetEnd) {
              patternIndex = lastMatch;
          }
      }

      return patternIndex == patternLength && offsetIndex == offsetEnd ;
  }

  /**
   * Compute:
   * - 100 * (sum(price * discount) / sum(price)) as discount_ratio
   * - (avg(price) filter(where discount = 0.05)) as avg_price
   */
  public static AggregatedBatch aggregate(FilteredBatch input) {
      AggregatedBatch output = new AggregatedBatch();
      output.discount_ratio  = new double[1];
      output.avg_price       = new double[1];

      Double resultAcc = 0d;
      Double price     = 0d;
      Double discount  = 0d;
      Double priceAcc  = 0d;

      int avgCnt      = 0;
      Double avgPrice = 0d;

      for (int rowId = 0; rowId < input.numRows; ++rowId) {
          price      = input.price[rowId];
          discount   = input.discount[rowId];
          priceAcc  += price;
          resultAcc += (price * discount);
          if (discount == 0.05) {
              avgPrice += price;
              ++avgCnt;
          }
      }

      output.discount_ratio[0] = 100 * (resultAcc / priceAcc); // Not catching exceptions for this.
      output.avg_price[0]      = avgPrice / avgCnt;
      output.numRows           = 1;
      return output;
  }

  private static void assertResult(double expected, double actual, String description) {
    if (Math.abs(expected - actual) > 0.000001d) {
      throw new RuntimeException("Unexpected " + description);
    }
  }


  private static void checkTest(boolean status, String description) {
    if (status) {
        System.out.printf("passed: %s%n", description);
    } else {
        System.out.printf("*FAILED: %s%n", description);
    }
  }
}