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.carbondata.sdk.file;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.UUID;
  
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.FileReader;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore;
  import org.apache.carbondata.core.metadata.converter.SchemaConverter;
  import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.datatype.Field;
  import org.apache.carbondata.core.metadata.datatype.StructField;
  import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
  import org.apache.carbondata.core.metadata.schema.table.TableSchema;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
  import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
  import org.apache.carbondata.core.reader.CarbonFooterReaderV3;
  import org.apache.carbondata.core.reader.CarbonHeaderReader;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.path.CarbonTablePath;
  import org.apache.carbondata.format.FileFooter3;
  import org.apache.carbondata.format.IndexHeader;
  import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
  import org.apache.carbondata.sdk.file.arrow.ArrowConverter;
  
  import org.apache.hadoop.conf.Configuration;
  
  /**
   * Schema reader for carbon files, including carbondata file, carbonindex file, and schema file
   */
  public class CarbonSchemaReader {
  
    /**
     * Read schema file and return the schema
     *
     * @param schemaFilePath complete path including schema file name
     * @return schema object
     * @throws IOException
     */
    @Deprecated
    public static Schema readSchemaInSchemaFile(String schemaFilePath) throws IOException {
      org.apache.carbondata.format.TableInfo tableInfo = CarbonUtil.readSchemaFile(schemaFilePath);
      SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
      List<ColumnSchema> schemaList = schemaConverter
          .fromExternalToWrapperTableInfo(tableInfo, "", "", "")
          .getFactTable()
          .getListOfColumns();
      return new Schema(schemaList);
    }
  
    /**
     * get carbondata/carbonindex file in path
     *
     * @param path      carbon file path
     * @param extension carbon file extension
     * @param conf      hadoop configuration support, can set s3a AK,SK,
     *                  end point and other conf with this
     * @return CarbonFile array
     */
    private static CarbonFile[] getCarbonFile(String path, final String extension, Configuration conf)
        throws IOException {
      String dataFilePath = path;
      if (!(dataFilePath.endsWith(extension))) {
        CarbonFile[] carbonFiles = FileFactory
            .getCarbonFile(path, conf)
            .listFiles(new CarbonFileFilter() {
              @Override
              public boolean accept(CarbonFile file) {
                if (file == null) {
                  return false;
                }
                return file.getName().endsWith(extension);
              }
            });
        if (carbonFiles == null || carbonFiles.length < 1) {
          throw new IOException("Carbon file not exists.");
        }
        return carbonFiles;
      } else {
        throw new CarbonDataLoadingException("Please ensure path "
            + path + " end with " + extension);
      }
    }
  
    /**
     * read schema from path,
     * path can be folder path, carbonindex file path, and carbondata file path
     * and will not check all files schema
     *
     * @param path file/folder path
     * @return schema
     * @throws IOException
     */
    public static Schema readSchema(String path) throws IOException {
      Configuration conf = new Configuration();
      return readSchema(path, false, conf);
    }
  
    /**
     * Converting carbon schema to arrow schema in byte[],
     * byte[] can be converted back arrowSchema by other arrow interface module like pyspark etc.
     *
     * @param path
     * @return
     * @throws IOException
     */
    public static byte[] getArrowSchemaAsBytes(String path) throws IOException {
      Schema schema = CarbonSchemaReader.readSchema(path).asOriginOrder();
      ArrowConverter arrowConverter = new ArrowConverter(schema, 0);
      final byte[] bytes = arrowConverter.toSerializeArray();
      return bytes;
    }
  
    /**
     * read schema from path,
     * path can be folder path, carbonindex file path, and carbondata file path
     * and will not check all files schema
     *
     * @param path file/folder path
     * @param conf hadoop configuration support, can set s3a AK,SK,end point and other conf with this
     * @return schema
     * @throws IOException
     */
    public static Schema readSchema(String path, Configuration conf) throws IOException {
      return readSchema(path, false, conf);
    }
  
    /**
     * read schema from path,
     * path can be folder path, carbonindex file path, and carbondata file path
     * and user can decide whether check all files schema
     *
     * @param path           file/folder path
     * @param validateSchema whether check all files schema
     * @param conf           hadoop configuration support, can set s3a AK,SK,
     *                       end point and other conf with this
     * @return schema
     * @throws IOException
     */
    public static Schema readSchema(String path, boolean validateSchema, Configuration conf)
        throws IOException {
      // Check whether it is transactional table reads the schema
      String schemaFilePath = CarbonTablePath.getSchemaFilePath(path, conf);
      if (FileFactory.getCarbonFile(schemaFilePath, conf).exists()) {
        return readSchemaInSchemaFile(schemaFilePath, conf);
      }
      if (path.endsWith(CarbonTablePath.INDEX_FILE_EXT)) {
        return readSchemaFromIndexFile(path, conf);
      } else if (path.endsWith(CarbonTablePath.CARBON_DATA_EXT)) {
        return readSchemaFromDataFile(path, conf);
      } else if (validateSchema) {
        CarbonFile[] carbonIndexFiles = getCarbonFile(path, CarbonTablePath.INDEX_FILE_EXT, conf);
        Schema schema;
        if (carbonIndexFiles != null && carbonIndexFiles.length != 0) {
          schema = readSchemaFromIndexFile(carbonIndexFiles[0].getAbsolutePath(), conf);
          for (int i = 1; i < carbonIndexFiles.length; i++) {
            Schema schema2 = readSchemaFromIndexFile(carbonIndexFiles[i].getAbsolutePath(), conf);
            if (!schema.equals(schema2)) {
              throw new CarbonDataLoadingException("Schema is different between different files.");
            }
          }
          CarbonFile[] carbonDataFiles = getCarbonFile(path, CarbonTablePath.CARBON_DATA_EXT, conf);
          for (int i = 0; i < carbonDataFiles.length; i++) {
            Schema schema2 = readSchemaFromDataFile(carbonDataFiles[i].getAbsolutePath(), conf);
            if (!schema.equals(schema2)) {
              throw new CarbonDataLoadingException("Schema is different between different files.");
            }
          }
          return schema;
        } else {
          throw new CarbonDataLoadingException("No carbonindex file in this path.");
        }
      } else {
        return readSchemaFromFolder(path, conf);
      }
    }
  
    private static Schema readSchemaInSchemaFile(String schemaFilePath, Configuration conf)
        throws IOException {
      org.apache.carbondata.format.TableInfo tableInfo =
          CarbonUtil.readSchemaFile(schemaFilePath, conf);
      SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
      TableSchema factTable =
          schemaConverter.fromExternalToWrapperTableInfo(tableInfo, "", "", "").getFactTable();
      List<ColumnSchema> schemaList = factTable.getListOfColumns();
      return new Schema(schemaList, factTable.getTableProperties());
    }
  
    /**
     * read schema from path,
     * path can be folder path, carbonindex file path, and carbondata file path
     * and user can decide whether check all files schema
     *
     * @param path           file/folder path
     * @param validateSchema whether check all files schema
     * @return schema
     * @throws IOException
     */
    public static Schema readSchema(String path, boolean validateSchema) throws IOException {
      Configuration conf = new Configuration();
      return readSchema(path, validateSchema, conf);
    }
  
    /**
     * Read carbondata file and return the schema
     * This interface will be removed,
     * please use readSchema instead of this interface
     *
     * @param dataFilePath carbondata file store path
     * @return Schema object
     * @throws IOException
     */
    @Deprecated
    public static Schema readSchemaInDataFile(String dataFilePath) throws IOException {
      return readSchema(dataFilePath, false);
    }
  
    /**
     * Read schema from carbondata file
     *
     * @param dataFilePath carbondata file path
     * @param conf         hadoop configuration support, can set s3a AK,SK,
     *                     end point and other conf with this
     * @return carbon data schema
     * @throws IOException
     */
    private static Schema readSchemaFromDataFile(String dataFilePath, Configuration conf)
        throws IOException {
      CarbonHeaderReader reader = new CarbonHeaderReader(dataFilePath, conf);
      List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();
      List<ColumnSchema> schemaList = reader.readSchema();
      for (int i = 0; i < schemaList.size(); i++) {
        ColumnSchema columnSchema = schemaList.get(i);
        if (!(columnSchema.isComplexColumn())) {
          columnSchemaList.add(columnSchema);
        }
      }
      return new Schema(columnSchemaList);
    }
  
    /**
     * Read carbonindex file and return the schema
     * This interface will be removed,
     * please use readSchema instead of this interface
     *
     * @param indexFilePath complete path including index file name
     * @return schema object
     * @throws IOException
     */
    @Deprecated
    public static Schema readSchemaInIndexFile(String indexFilePath) throws IOException {
      return readSchema(indexFilePath, false);
    }
  
    public static List<StructField> getChildrenCommon(CarbonTable table, String columnName) {
      List<CarbonDimension> list = table.getChildren(columnName);
      List<StructField> structFields = new ArrayList<StructField>();
      for (int i = 0; i < list.size(); i++) {
        CarbonDimension carbonDimension = list.get(i);
        if (DataTypes.isStructType(carbonDimension.getDataType())) {
          structFields.add(getStructChildren(table, carbonDimension.getColName()));
          return structFields;
        } else if (DataTypes.isArrayType(carbonDimension.getDataType())) {
          structFields.add(getArrayChildren(table, carbonDimension.getColName()));
          return structFields;
        } else if (DataTypes.isMapType(carbonDimension.getDataType())) {
          //TODO
        } else {
          ColumnSchema columnSchema = carbonDimension.getColumnSchema();
          structFields.add(new StructField(columnSchema.getColumnName(), columnSchema.getDataType()));
        }
      }
      return structFields;
    }
  
    public static StructField getStructChildren(CarbonTable table, String columnName) {
      List<StructField> structFields = getChildrenCommon(table, columnName);
      return new StructField(columnName, DataTypes.createStructType(structFields));
    }
  
    public static StructField getArrayChildren(CarbonTable table, String columnName) {
      List<StructField> structFields = getChildrenCommon(table, columnName);
      return structFields.get(0);
    }
  
    /**
     * Read schema from carbon file folder path
     *
     * @param folderPath carbon file folder path
     * @param conf       hadoop configuration support, can set s3a AK,SK,
     *                   end point and other conf with this
     * @return carbon data Schema
     * @throws IOException
     */
    private static Schema readSchemaFromFolder(String folderPath, Configuration conf)
      throws IOException {
      String tableName = "UnknownTable" + UUID.randomUUID();
      CarbonTable table = CarbonTable.buildTable(folderPath, tableName, conf);
      List<ColumnSchema> columnSchemaList = table.getTableInfo().getFactTable().getListOfColumns();
      int numOfChildren = 0;
      for (ColumnSchema columnSchema : columnSchemaList) {
        if (!(columnSchema.getColumnName().contains(CarbonCommonConstants.POINT))) {
          numOfChildren++;
        }
      }
      Field[] fields = new Field[numOfChildren];
  
      int indexOfFields = 0;
      for (ColumnSchema columnSchema : columnSchemaList) {
        if (!columnSchema.getColumnName().contains(CarbonCommonConstants.POINT)) {
          if (DataTypes.isStructType(columnSchema.getDataType())) {
            StructField structField = getStructChildren(table, columnSchema.getColumnName());
            List<StructField> list = new ArrayList<>();
            list.add(structField);
            fields[indexOfFields] = new Field(columnSchema.getColumnName(),
              DataTypes.createStructType(list));
            fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());
            indexOfFields++;
          } else if (DataTypes.isArrayType(columnSchema.getDataType())) {
            StructField structField = getArrayChildren(table, columnSchema.getColumnName());
            List<StructField> list = new ArrayList<>();
            list.add(structField);
            fields[indexOfFields] = new Field(columnSchema.getColumnName(), "array", list);
            fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());
            indexOfFields++;
          } else if (DataTypes.isMapType(columnSchema.getDataType())) {
            //TODO
          } else {
            fields[indexOfFields] = new Field(columnSchema);
            fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());
            indexOfFields++;
          }
        }
      }
      return new Schema(fields);
    }
  
    /**
     * Read schema from carbonindex file
     *
     * @param indexFilePath carbonindex file path
     * @param conf          hadoop configuration support, can set s3a AK,SK,
     *                      end point and other conf with this
     * @return carbon data Schema
     * @throws IOException
     */
    private static Schema readSchemaFromIndexFile(String indexFilePath, Configuration conf) {
      IndexHeader readIndexHeader = SegmentIndexFileStore.readIndexHeader(indexFilePath, conf);
      List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();
      List<org.apache.carbondata.format.ColumnSchema> tableColumns =
          readIndexHeader.getTable_columns();
      for (org.apache.carbondata.format.ColumnSchema columnSchema : tableColumns) {
        if (!(columnSchema.column_name.contains("."))) {
          columnSchemaList.add(CarbonUtil.thriftColumnSchemaToWrapperColumnSchema(columnSchema));
        }
      }
      return new Schema(columnSchemaList);
    }
  
    /**
     * This method return the version details in formatted string by reading from carbondata file
     *
     * @param dataFilePath
     * @return
     * @throws IOException
     */
    public static String getVersionDetails(String dataFilePath) throws IOException {
      long fileSize =
          FileFactory.getCarbonFile(dataFilePath).getSize();
      FileReader fileReader = FileFactory.getFileHolder(FileFactory.getFileType(dataFilePath));
      ByteBuffer buffer =
          fileReader.readByteBuffer(FileFactory.getUpdatedFilePath(dataFilePath), fileSize - 8, 8);
      fileReader.finish();
      CarbonFooterReaderV3 footerReader =
          new CarbonFooterReaderV3(dataFilePath, buffer.getLong());
      FileFooter3 footer = footerReader.readFooterVersion3();
      if (null != footer.getExtra_info()) {
        return footer.getExtra_info().get(CarbonCommonConstants.CARBON_WRITTEN_BY_FOOTER_INFO)
            + " in version: " + footer.getExtra_info()
            .get(CarbonCommonConstants.CARBON_WRITTEN_VERSION);
      } else {
        return "Version Details are not found in carbondata file";
      }
    }
  }
  