3862

Build # 3862

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/b165e127bdbd4214ef2f00bd20625d6b0343ff4e">b165e127b<a href="https://github.com/apache/carbondata/commit/b165e127bdbd4214ef2f00bd20625d6b0343ff4e">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/208b495587f1557c5520d22da23dfd99dbdc8966">208b49558</a>">208b49558</a><a href="https://github.com/apache/carbondata/commit/b165e127bdbd4214ef2f00bd20625d6b0343ff4e"> into fc8c9d067">fc8c9d067</a>

Run Details

72 of 72 new or added lines in 6 files covered. (100.0%)

64145 of 82955 relevant lines covered (77.33%)

1.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.67

/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonSchemaReader.java

/*
 * 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.DataInputStream;
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.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.StructField;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
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.reader.CarbonIndexFileReader;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.format.FileFooter3;
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
import org.apache.carbondata.sdk.file.arrow.ArrowConverter;

import static org.apache.carbondata.core.util.CarbonUtil.thriftColumnSchemaToWrapperColumnSchema;
import static org.apache.carbondata.core.util.path.CarbonTablePath.CARBON_DATA_EXT;
import static org.apache.carbondata.core.util.path.CarbonTablePath.INDEX_FILE_EXT;

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 {
    if (path.endsWith(INDEX_FILE_EXT)) {
      return readSchemaFromIndexFile(path, conf);
    } else if (path.endsWith(CARBON_DATA_EXT)) {
      return readSchemaFromDataFile(path, conf);
    } else if (validateSchema) {
      CarbonFile[] carbonIndexFiles = getCarbonFile(path, 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, 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);
    }
  }

  /**
   * 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.getColumnName().contains("."))) {
        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 StructField getStructChildren(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())) {
        return getStructChildren(table, carbonDimension.getColName());
      } else if (DataTypes.isArrayType(carbonDimension.getDataType())) {
        return getArrayChildren(table, carbonDimension.getColName());
      } else if (DataTypes.isMapType(carbonDimension.getDataType())) {
        //TODO
      } else {
        ColumnSchema columnSchema = carbonDimension.getColumnSchema();
        structFields.add(new StructField(columnSchema.getColumnName(), columnSchema.getDataType()));
      }
    }
    return new StructField(columnName, DataTypes.createStructType(structFields));
  }

  public static StructField getArrayChildren(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())) {
        return getStructChildren(table, carbonDimension.getColName());
      } else if (DataTypes.isArrayType(carbonDimension.getDataType())) {
        return getArrayChildren(table, carbonDimension.getColName());
      } else if (DataTypes.isMapType(carbonDimension.getDataType())) {
        //TODO
      } else {
        ColumnSchema columnSchema = carbonDimension.getColumnSchema();
        structFields.add(new StructField(columnSchema.getColumnName(), columnSchema.getDataType()));
      }
    }
    return structFields.get(0);
  }

  /**
   * 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)
      throws IOException {
    CarbonFile indexFile =
        FileFactory.getCarbonFile(indexFilePath, conf);
    if (!indexFile.getName().endsWith(INDEX_FILE_EXT)) {
      throw new IOException("Not an index file name");
    }
    // read schema from the first index file
    DataInputStream dataInputStream =
        FileFactory.getDataInputStream(indexFilePath, FileFactory.getFileType(indexFilePath), conf);
    byte[] bytes = new byte[(int) indexFile.getSize()];
    try {
      //get the file in byte buffer
      dataInputStream.readFully(bytes);
      CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
      // read from byte buffer.
      indexReader.openThriftReader(bytes);
      // get the index header
      org.apache.carbondata.format.IndexHeader readIndexHeader = indexReader.readIndexHeader();
      List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();
      List<org.apache.carbondata.format.ColumnSchema> table_columns =
          readIndexHeader.getTable_columns();
      for (org.apache.carbondata.format.ColumnSchema columnSchema : table_columns) {
        if (!(columnSchema.column_name.contains("."))) {
          columnSchemaList.add(thriftColumnSchemaToWrapperColumnSchema(columnSchema));
        }
      }
      return new Schema(columnSchemaList);
    } finally {
      dataInputStream.close();
    }
  }


  /**
   * 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 sum = 0;
    for (ColumnSchema columnSchema : columnSchemaList) {
      if (!(columnSchema.getColumnName().contains("."))) {
        sum++;
      }
    }
    Field[] fields = new Field[sum];

    int indexOfFields = 0;
    for (ColumnSchema columnSchema : table.getTableInfo().getFactTable().getListOfColumns()) {
      if (!columnSchema.getColumnName().contains(".")) {
        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);
  }

  /**
   * 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, FileFactory.getFileType(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";
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.sdk.file;
19
20	import java.io.DataInputStream;
21	import java.io.IOException;
22	import java.nio.ByteBuffer;
23	import java.util.ArrayList;
24	import java.util.List;
25	import java.util.UUID;
26
27	import org.apache.carbondata.core.constants.CarbonCommonConstants;
28	import org.apache.carbondata.core.datastore.FileReader;
29	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
30	import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
31	import org.apache.carbondata.core.datastore.impl.FileFactory;
32	import org.apache.carbondata.core.metadata.converter.SchemaConverter;
33	import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
34	import org.apache.carbondata.core.metadata.datatype.DataTypes;
35	import org.apache.carbondata.core.metadata.datatype.StructField;
36	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
37	import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
38	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
39	import org.apache.carbondata.core.reader.CarbonFooterReaderV3;
40	import org.apache.carbondata.core.reader.CarbonHeaderReader;
41	import org.apache.carbondata.core.reader.CarbonIndexFileReader;
42	import org.apache.carbondata.core.util.CarbonUtil;
43	import org.apache.carbondata.format.FileFooter3;
44	import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
45	import org.apache.carbondata.sdk.file.arrow.ArrowConverter;
46
47	import static org.apache.carbondata.core.util.CarbonUtil.thriftColumnSchemaToWrapperColumnSchema;
48	import static org.apache.carbondata.core.util.path.CarbonTablePath.CARBON_DATA_EXT;
49	import static org.apache.carbondata.core.util.path.CarbonTablePath.INDEX_FILE_EXT;
50
51	import org.apache.hadoop.conf.Configuration;
52
53	/**
54	* Schema reader for carbon files, including carbondata file, carbonindex file, and schema file
55	*/
56	public class CarbonSchemaReader {	×
57
58	/**
59	* Read schema file and return the schema
60	*
61	* @param schemaFilePath complete path including schema file name
62	* @return schema object
63	* @throws IOException
64	*/
65	@Deprecated
66	public static Schema readSchemaInSchemaFile(String schemaFilePath) throws IOException {
67	org.apache.carbondata.format.TableInfo tableInfo = CarbonUtil.readSchemaFile(schemaFilePath);	×
68	SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();	×
69	List<ColumnSchema> schemaList = schemaConverter	×
70	.fromExternalToWrapperTableInfo(tableInfo, "", "", "")	×
71	.getFactTable()	×
72	.getListOfColumns();	×
73	return new Schema(schemaList);	×
74	}
75
76	/**
77	* get carbondata/carbonindex file in path
78	*
79	* @param path carbon file path
80	* @param extension carbon file extension
81	* @param conf hadoop configuration support, can set s3a AK,SK,
82	* end point and other conf with this
83	* @return CarbonFile array
84	*/
85	private static CarbonFile[] getCarbonFile(String path, final String extension, Configuration conf)
86	throws IOException {
87	String dataFilePath = path;	2✔
88	if (!(dataFilePath.endsWith(extension))) {	2✔
89	CarbonFile[] carbonFiles = FileFactory	2✔
90	.getCarbonFile(path, conf)	2✔
91	.listFiles(new CarbonFileFilter() {	2✔
92	@Override
93	public boolean accept(CarbonFile file) {
94	if (file == null) {	2✔
95	return false;	×
96	}
97	return file.getName().endsWith(extension);	2✔
98	}
99	});
100	if (carbonFiles == null \|\| carbonFiles.length < 1) {	2✔
101	throw new IOException("Carbon file not exists.");	×
102	}
103	return carbonFiles;	2✔
104	} else {
105	throw new CarbonDataLoadingException("Please ensure path "	×
106	+ path + " end with " + extension);
107	}
108	}
109
110	/**
111	* read schema from path,
112	* path can be folder path, carbonindex file path, and carbondata file path
113	* and will not check all files schema
114	*
115	* @param path file/folder path
116	* @return schema
117	* @throws IOException
118	*/
119	public static Schema readSchema(String path) throws IOException {
120	Configuration conf = new Configuration();	2✔
121	return readSchema(path, false, conf);	2✔
122	}
123
124	/**
125	* Converting carbon schema to arrow schema in byte[],
126	* byte[] can be converted back arrowSchema by other arrow interface module like pyspark etc.
127	*
128	* @param path
129	* @return
130	* @throws IOException
131	*/
132	public static byte[] getArrowSchemaAsBytes(String path) throws IOException {
133	Schema schema = CarbonSchemaReader.readSchema(path).asOriginOrder();	2✔
134	ArrowConverter arrowConverter = new ArrowConverter(schema, 0);	2✔
135	final byte[] bytes = arrowConverter.toSerializeArray();	2✔
136	return bytes;	2✔
137	}
138
139	/**
140	* read schema from path,
141	* path can be folder path, carbonindex file path, and carbondata file path
142	* and will not check all files schema
143	*
144	* @param path file/folder path
145	* @param conf hadoop configuration support, can set s3a AK,SK,end point and other conf with this
146	* @return schema
147	* @throws IOException
148	*/
149	public static Schema readSchema(String path, Configuration conf) throws IOException {
150	return readSchema(path, false, conf);	×
151	}
152
153	/**
154	* read schema from path,
155	* path can be folder path, carbonindex file path, and carbondata file path
156	* and user can decide whether check all files schema
157	*
158	* @param path file/folder path
159	* @param validateSchema whether check all files schema
160	* @param conf hadoop configuration support, can set s3a AK,SK,
161	* end point and other conf with this
162	* @return schema
163	* @throws IOException
164	*/
165	public static Schema readSchema(String path, boolean validateSchema, Configuration conf)
166	throws IOException {
167	if (path.endsWith(INDEX_FILE_EXT)) {	2✔
168	return readSchemaFromIndexFile(path, conf);	2✔
169	} else if (path.endsWith(CARBON_DATA_EXT)) {	2✔
170	return readSchemaFromDataFile(path, conf);	2✔
171	} else if (validateSchema) {	2✔
172	CarbonFile[] carbonIndexFiles = getCarbonFile(path, INDEX_FILE_EXT, conf);	2✔
173	Schema schema;
174	if (carbonIndexFiles != null && carbonIndexFiles.length != 0) {	2✔
175	schema = readSchemaFromIndexFile(carbonIndexFiles[0].getAbsolutePath(), conf);	2✔
176	for (int i = 1; i < carbonIndexFiles.length; i++) {	2✔
177	Schema schema2 = readSchemaFromIndexFile(carbonIndexFiles[i].getAbsolutePath(), conf);	2✔
178	if (!schema.equals(schema2)) {	2✔
179	throw new CarbonDataLoadingException("Schema is different between different files.");	2✔
180	}
181	}
182	CarbonFile[] carbonDataFiles = getCarbonFile(path, CARBON_DATA_EXT, conf);	2✔
183	for (int i = 0; i < carbonDataFiles.length; i++) {	2✔
184	Schema schema2 = readSchemaFromDataFile(carbonDataFiles[i].getAbsolutePath(), conf);	2✔
185	if (!schema.equals(schema2)) {	2✔
186	throw new CarbonDataLoadingException("Schema is different between different files.");	×
187	}
188	}
189	return schema;	2✔
190	} else {
191	throw new CarbonDataLoadingException("No carbonindex file in this path.");	×
192	}
193	} else {
194	return readSchemaFromFolder(path, conf);	2✔
195	}
196	}
197
198	/**
199	* read schema from path,
200	* path can be folder path, carbonindex file path, and carbondata file path
201	* and user can decide whether check all files schema
202	*
203	* @param path file/folder path
204	* @param validateSchema whether check all files schema
205	* @return schema
206	* @throws IOException
207	*/
208	public static Schema readSchema(String path, boolean validateSchema) throws IOException {
209	Configuration conf = new Configuration();	2✔
210	return readSchema(path, validateSchema, conf);	2✔
211	}
212
213	/**
214	* Read carbondata file and return the schema
215	* This interface will be removed,
216	* please use readSchema instead of this interface
217	*
218	* @param dataFilePath carbondata file store path
219	* @return Schema object
220	* @throws IOException
221	*/
222	@Deprecated
223	public static Schema readSchemaInDataFile(String dataFilePath) throws IOException {
224	return readSchema(dataFilePath, false);	×
225	}
226
227	/**
228	* Read schema from carbondata file
229	*
230	* @param dataFilePath carbondata file path
231	* @param conf hadoop configuration support, can set s3a AK,SK,
232	* end point and other conf with this
233	* @return carbon data schema
234	* @throws IOException
235	*/
236	private static Schema readSchemaFromDataFile(String dataFilePath, Configuration conf)
237	throws IOException {
238	CarbonHeaderReader reader = new CarbonHeaderReader(dataFilePath, conf);	2✔
239	List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();	2✔
240	List<ColumnSchema> schemaList = reader.readSchema();	2✔
241	for (int i = 0; i < schemaList.size(); i++) {	2✔
242	ColumnSchema columnSchema = schemaList.get(i);	2✔
243	if (!(columnSchema.getColumnName().contains("."))) {	2✔
244	columnSchemaList.add(columnSchema);	2✔
245	}
246	}
247	return new Schema(columnSchemaList);	2✔
248	}
249
250	/**
251	* Read carbonindex file and return the schema
252	* This interface will be removed,
253	* please use readSchema instead of this interface
254	*
255	* @param indexFilePath complete path including index file name
256	* @return schema object
257	* @throws IOException
258	*/
259	@Deprecated
260	public static Schema readSchemaInIndexFile(String indexFilePath) throws IOException {
261	return readSchema(indexFilePath, false);	×
262	}
263
264	public static StructField getStructChildren(CarbonTable table, String columnName) {
265	List<CarbonDimension> list = table.getChildren(columnName);	2✔
266	List<StructField> structFields = new ArrayList<StructField>();	2✔
267	for (int i = 0; i < list.size(); i++) {	2✔
268	CarbonDimension carbonDimension = list.get(i);	2✔
269	if (DataTypes.isStructType(carbonDimension.getDataType())) {	2✔
270	return getStructChildren(table, carbonDimension.getColName());	×
271	} else if (DataTypes.isArrayType(carbonDimension.getDataType())) {	2✔
272	return getArrayChildren(table, carbonDimension.getColName());	×
273	} else if (DataTypes.isMapType(carbonDimension.getDataType())) {	2✔
274	//TODO
275	} else {
276	ColumnSchema columnSchema = carbonDimension.getColumnSchema();	2✔
277	structFields.add(new StructField(columnSchema.getColumnName(), columnSchema.getDataType()));	2✔
278	}
279	}
280	return new StructField(columnName, DataTypes.createStructType(structFields));	2✔
281	}
282
283	public static StructField getArrayChildren(CarbonTable table, String columnName) {
284	List<CarbonDimension> list = table.getChildren(columnName);	2✔
285	List<StructField> structFields = new ArrayList<StructField>();	2✔
286	for (int i = 0; i < list.size(); i++) {	2✔
287	CarbonDimension carbonDimension = list.get(i);	2✔
288	if (DataTypes.isStructType(carbonDimension.getDataType())) {	2✔
289	return getStructChildren(table, carbonDimension.getColName());	2✔
290	} else if (DataTypes.isArrayType(carbonDimension.getDataType())) {	2✔
291	return getArrayChildren(table, carbonDimension.getColName());	×
292	} else if (DataTypes.isMapType(carbonDimension.getDataType())) {	2✔
293	//TODO
294	} else {
295	ColumnSchema columnSchema = carbonDimension.getColumnSchema();	2✔
296	structFields.add(new StructField(columnSchema.getColumnName(), columnSchema.getDataType()));	2✔
297	}
298	}
299	return structFields.get(0);	2✔
300	}
301
302	/**
303	* Read schema from carbonindex file
304	*
305	* @param indexFilePath carbonindex file path
306	* @param conf hadoop configuration support, can set s3a AK,SK,
307	* end point and other conf with this
308	* @return carbon data Schema
309	* @throws IOException
310	*/
311	private static Schema readSchemaFromIndexFile(String indexFilePath, Configuration conf)
312	throws IOException {
313	CarbonFile indexFile =	2✔
314	FileFactory.getCarbonFile(indexFilePath, conf);	2✔
315	if (!indexFile.getName().endsWith(INDEX_FILE_EXT)) {	2✔
316	throw new IOException("Not an index file name");	×
317	}
318	// read schema from the first index file
319	DataInputStream dataInputStream =	2✔
320	FileFactory.getDataInputStream(indexFilePath, FileFactory.getFileType(indexFilePath), conf);	2✔
321	byte[] bytes = new byte[(int) indexFile.getSize()];	2✔
322	try {
323	//get the file in byte buffer
324	dataInputStream.readFully(bytes);	2✔
325	CarbonIndexFileReader indexReader = new CarbonIndexFileReader();	2✔
326	// read from byte buffer.
327	indexReader.openThriftReader(bytes);	2✔
328	// get the index header
329	org.apache.carbondata.format.IndexHeader readIndexHeader = indexReader.readIndexHeader();	2✔
330	List<ColumnSchema> columnSchemaList = new ArrayList<ColumnSchema>();	2✔
331	List<org.apache.carbondata.format.ColumnSchema> table_columns =	2✔
332	readIndexHeader.getTable_columns();	2✔
333	for (org.apache.carbondata.format.ColumnSchema columnSchema : table_columns) {	2✔
334	if (!(columnSchema.column_name.contains("."))) {	2✔
335	columnSchemaList.add(thriftColumnSchemaToWrapperColumnSchema(columnSchema));	2✔
336	}
337	}	2✔
338	return new Schema(columnSchemaList);	2✔
339	} finally {
340	dataInputStream.close();	2✔
341	}	×
342	}
343
344
345	/**
346	* Read schema from carbon file folder path
347	*
348	* @param folderPath carbon file folder path
349	* @param conf hadoop configuration support, can set s3a AK,SK,
350	* end point and other conf with this
351	* @return carbon data Schema
352	* @throws IOException
353	*/
354	private static Schema readSchemaFromFolder(String folderPath, Configuration conf)
355	throws IOException {
356	String tableName = "UnknownTable" + UUID.randomUUID();	2✔
357	CarbonTable table = CarbonTable.buildTable(folderPath, tableName, conf);	2✔
358	List<ColumnSchema> columnSchemaList = table.getTableInfo().getFactTable().getListOfColumns();	2✔
359	int sum = 0;	2✔
360	for (ColumnSchema columnSchema : columnSchemaList) {	2✔
361	if (!(columnSchema.getColumnName().contains("."))) {	2✔
362	sum++;	2✔
363	}
364	}	2✔
365	Field[] fields = new Field[sum];	2✔
366
367	int indexOfFields = 0;	2✔
368	for (ColumnSchema columnSchema : table.getTableInfo().getFactTable().getListOfColumns()) {	2✔
369	if (!columnSchema.getColumnName().contains(".")) {	2✔
370	if (DataTypes.isStructType(columnSchema.getDataType())) {	2✔
371	StructField structField = getStructChildren(table, columnSchema.getColumnName());	×
372	List<StructField> list = new ArrayList<>();	×
373	list.add(structField);	×
374	fields[indexOfFields] = new Field(columnSchema.getColumnName(),	×
375	DataTypes.createStructType(list));	×
376	fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());	×
377	indexOfFields++;	×
378	} else if (DataTypes.isArrayType(columnSchema.getDataType())) {	2✔
379	StructField structField = getArrayChildren(table, columnSchema.getColumnName());	2✔
380	List<StructField> list = new ArrayList<>();	2✔
381	list.add(structField);	2✔
382	fields[indexOfFields] = new Field(columnSchema.getColumnName(), "array", list);	2✔
383	fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());	2✔
384	indexOfFields++;	2✔
385	} else if (DataTypes.isMapType(columnSchema.getDataType())) {	2✔
386	//TODO
387	} else {
388	fields[indexOfFields] = new Field(columnSchema);	2✔
389	fields[indexOfFields].setSchemaOrdinal(columnSchema.getSchemaOrdinal());	2✔
390	indexOfFields++;	2✔
391	}
392	}
393	}	2✔
394	return new Schema(fields);	2✔
395	}
396
397	/**
398	* This method return the version details in formatted string by reading from carbondata file
399	*
400	* @param dataFilePath
401	* @return
402	* @throws IOException
403	*/
404	public static String getVersionDetails(String dataFilePath) throws IOException {
405	long fileSize =	2✔
406	FileFactory.getCarbonFile(dataFilePath, FileFactory.getFileType(dataFilePath)).getSize();	2✔
407	FileReader fileReader = FileFactory.getFileHolder(FileFactory.getFileType(dataFilePath));	2✔
408	ByteBuffer buffer =	2✔
409	fileReader.readByteBuffer(FileFactory.getUpdatedFilePath(dataFilePath), fileSize - 8, 8);	2✔
410	fileReader.finish();	2✔
411	CarbonFooterReaderV3 footerReader = new CarbonFooterReaderV3(dataFilePath, buffer.getLong());	2✔
412	FileFooter3 footer = footerReader.readFooterVersion3();	2✔
413	if (null != footer.getExtra_info()) {	2✔
414	return footer.getExtra_info().get(CarbonCommonConstants.CARBON_WRITTEN_BY_FOOTER_INFO)	2✔
415	+ " in version: " + footer.getExtra_info()	2✔
416	.get(CarbonCommonConstants.CARBON_WRITTEN_VERSION);	2✔
417	} else {
418	return "Version Details are not found in carbondata file";	×
419	}
420	}
421	}

apache / carbondata / 3862

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous