content: /*
   * 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.beam.sdk.tpcds;
  
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  
  /** Generate the tpcds queries execution summary on the command line after finishing all jobs. */
  public class SummaryGenerator {
    private static final int PADDING_SIZE = 2;
    private static final String NEW_LINE = "\n";
    private static final String TABLE_JOINT_SYMBOL = "+";
    private static final String TABLE_V_SPLIT_SYMBOL = "|";
    private static final String TABLE_H_SPLIT_SYMBOL = "-";
  
    public static String generateTable(
        List<String> headersList, List<List<String>> rowsList, int... overRiddenHeaderHeight) {
      StringBuilder stringBuilder = new StringBuilder();
  
      int rowHeight = overRiddenHeaderHeight.length > 0 ? overRiddenHeaderHeight[0] : 1;
  
      Map<Integer, Integer> columnMaxWidthMapping = getMaximumWidthofTable(headersList, rowsList);
  
      stringBuilder.append(NEW_LINE);
      stringBuilder.append(NEW_LINE);
      createRowLine(stringBuilder, headersList.size(), columnMaxWidthMapping);
      stringBuilder.append(NEW_LINE);
  
      for (int headerIndex = 0; headerIndex < headersList.size(); headerIndex++) {
        fillCell(stringBuilder, headersList.get(headerIndex), headerIndex, columnMaxWidthMapping);
      }
  
      stringBuilder.append(NEW_LINE);
  
      createRowLine(stringBuilder, headersList.size(), columnMaxWidthMapping);
  
      for (List<String> row : rowsList) {
        for (int i = 0; i < rowHeight; i++) {
          stringBuilder.append(NEW_LINE);
        }
        for (int cellIndex = 0; cellIndex < row.size(); cellIndex++) {
          fillCell(stringBuilder, row.get(cellIndex), cellIndex, columnMaxWidthMapping);
        }
      }
  
      stringBuilder.append(NEW_LINE);
      createRowLine(stringBuilder, headersList.size(), columnMaxWidthMapping);
      stringBuilder.append(NEW_LINE);
      stringBuilder.append(NEW_LINE);
  
      return stringBuilder.toString();
    }
  
    private static void fillSpace(StringBuilder stringBuilder, int length) {
      for (int i = 0; i < length; i++) {
        stringBuilder.append(" ");
      }
    }
  
    /**
     * Add a rowLine at the beginning, the middle between headersList and rowLists, the end of the
     * summary table.
     */
    private static void createRowLine(
        StringBuilder stringBuilder,
        int headersListSize,
        Map<Integer, Integer> columnMaxWidthMapping) {
      for (int i = 0; i < headersListSize; i++) {
        if (i == 0) {
          stringBuilder.append(TABLE_JOINT_SYMBOL);
        }
  
        int columnMaxWidth = Optional.ofNullable(columnMaxWidthMapping.get(i)).orElse(0);
        for (int j = 0; j < columnMaxWidth + PADDING_SIZE * 2; j++) {
          stringBuilder.append(TABLE_H_SPLIT_SYMBOL);
        }
        stringBuilder.append(TABLE_JOINT_SYMBOL);
      }
    }
  
    /** Get the width of the summary table. */
    private static Map<Integer, Integer> getMaximumWidthofTable(
        List<String> headersList, List<List<String>> rowsList) {
      Map<Integer, Integer> columnMaxWidthMapping = new HashMap<>();
  
      for (int columnIndex = 0; columnIndex < headersList.size(); columnIndex++) {
        columnMaxWidthMapping.put(columnIndex, 0);
      }
  
      for (int columnIndex = 0; columnIndex < headersList.size(); columnIndex++) {
        Integer columnMaxWidth =
            Optional.ofNullable(columnMaxWidthMapping.get(columnIndex)).orElse(0);
        if (headersList.get(columnIndex).length() > columnMaxWidth) {
          columnMaxWidthMapping.put(columnIndex, headersList.get(columnIndex).length());
        }
      }
  
      for (List<String> row : rowsList) {
        for (int columnIndex = 0; columnIndex < row.size(); columnIndex++) {
          Integer columnMaxWidth =
              Optional.ofNullable(columnMaxWidthMapping.get(columnIndex)).orElse(0);
          if (row.get(columnIndex).length() > columnMaxWidth) {
            columnMaxWidthMapping.put(columnIndex, row.get(columnIndex).length());
          }
        }
      }
  
      for (int columnIndex = 0; columnIndex < headersList.size(); columnIndex++) {
        int columnMaxWidth = Optional.ofNullable(columnMaxWidthMapping.get(columnIndex)).orElse(0);
        if (columnMaxWidth % 2 != 0) {
          columnMaxWidthMapping.put(columnIndex, columnMaxWidth + 1);
        }
      }
  
      return columnMaxWidthMapping;
    }
  
    private static int getOptimumCellPadding(
        int cellIndex,
        int datalength,
        Map<Integer, Integer> columnMaxWidthMapping,
        int cellPaddingSize) {
      if (datalength % 2 != 0) {
        datalength++;
      }
  
      int columnMaxWidth = Optional.ofNullable(columnMaxWidthMapping.get(cellIndex)).orElse(0);
      if (datalength < columnMaxWidth) {
        cellPaddingSize = cellPaddingSize + (columnMaxWidth - datalength) / 2;
      }
  
      return cellPaddingSize;
    }
  
    /** Use space to fill a single cell with optimum cell padding size. */
    private static void fillCell(
        StringBuilder stringBuilder,
        String cell,
        int cellIndex,
        Map<Integer, Integer> columnMaxWidthMapping) {
  
      int cellPaddingSize =
          getOptimumCellPadding(cellIndex, cell.length(), columnMaxWidthMapping, PADDING_SIZE);
  
      if (cellIndex == 0) {
        stringBuilder.append(TABLE_V_SPLIT_SYMBOL);
      }
  
      fillSpace(stringBuilder, cellPaddingSize);
      stringBuilder.append(cell);
      if (cell.length() % 2 != 0) {
        stringBuilder.append(" ");
      }
  
      fillSpace(stringBuilder, cellPaddingSize);
  
      stringBuilder.append(TABLE_V_SPLIT_SYMBOL);
    }
  }
  