| /* |
| * 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.drill.exec.physical.impl.writer; |
| |
| import org.apache.calcite.util.Pair; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.drill.categories.ParquetTest; |
| import org.apache.drill.categories.SlowTest; |
| import org.apache.drill.categories.UnlikelyTest; |
| import org.apache.drill.common.logical.FormatPluginConfig; |
| import org.apache.drill.common.util.DrillVersionInfo; |
| import org.apache.drill.exec.ExecConstants; |
| import org.apache.drill.exec.fn.interp.TestConstantFolding; |
| import org.apache.drill.exec.planner.physical.PlannerSettings; |
| import org.apache.drill.exec.store.dfs.FileSystemConfig; |
| import org.apache.drill.exec.store.parquet.ParquetFormatConfig; |
| import org.apache.drill.exec.util.JsonStringArrayList; |
| import com.google.common.base.Joiner; |
| import com.google.common.collect.ImmutableList; |
| import org.apache.drill.test.ClusterFixture; |
| import org.apache.drill.test.ClusterTest; |
| import org.apache.drill.test.TestBuilder; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.parquet.hadoop.ParquetFileReader; |
| import org.apache.parquet.hadoop.metadata.ParquetMetadata; |
| import org.apache.parquet.schema.MessageType; |
| import org.apache.parquet.schema.LogicalTypeAnnotation; |
| import org.apache.parquet.schema.PrimitiveType; |
| import org.joda.time.Period; |
| |
| import org.junit.AfterClass; |
| import org.junit.BeforeClass; |
| import org.junit.Ignore; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| |
| import java.io.File; |
| import java.io.FileWriter; |
| import java.io.IOException; |
| import java.math.BigDecimal; |
| import java.nio.charset.Charset; |
| import java.nio.file.Paths; |
| import java.time.LocalDate; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import static java.util.Arrays.asList; |
| import static org.apache.drill.exec.store.parquet.ParquetRecordWriter.DRILL_VERSION_PROPERTY; |
| import static org.apache.drill.exec.util.StoragePluginTestUtils.DFS_TMP_SCHEMA; |
| import static org.apache.drill.test.TestBuilder.convertToLocalDateTime; |
| import static org.apache.drill.test.TestBuilder.mapOf; |
| import static org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER; |
| import static org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GROUPS; |
| import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; |
| import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY; |
| import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; |
| import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; |
| import static org.junit.jupiter.api.Assertions.assertEquals; |
| |
| @RunWith(Parameterized.class) |
| @Category({SlowTest.class, ParquetTest.class}) |
| public class TestParquetWriter extends ClusterTest { |
| @Parameterized.Parameters |
| public static Collection<Object[]> data() { |
| return Arrays.asList(new Object[][] { {100} }); |
| } |
| |
| @BeforeClass |
| public static void setupTestFiles() { |
| dirTestWatcher.copyResourceToRoot(Paths.get("parquet", "int96_dict_change")); |
| } |
| |
| // Map storing a convenient name as well as the cast type necessary |
| // to produce it casting from a varchar |
| private static final Map<String, String> allTypes = new HashMap<>(); |
| |
| // Select statement for all supported Drill types, for use in conjunction with |
| // the file parquet/alltypes.json in the resources directory |
| private static final String allTypesSelection; |
| |
| static { |
| allTypes.put("int", "int"); |
| allTypes.put("bigint", "bigint"); |
| allTypes.put("decimal(9, 4)", "decimal9"); |
| allTypes.put("decimal(18,9)", "decimal18"); |
| allTypes.put("decimal(28, 14)", "decimal28sparse"); |
| allTypes.put("decimal(38, 19)", "decimal38sparse"); |
| allTypes.put("decimal(38, 15)", "vardecimal"); |
| allTypes.put("date", "date"); |
| allTypes.put("timestamp", "timestamp"); |
| allTypes.put("float", "float4"); |
| allTypes.put("double", "float8"); |
| allTypes.put("varbinary(65000)", "varbinary"); |
| // TODO(DRILL-2297) |
| // allTypes.put("interval year", "intervalyear"); |
| allTypes.put("interval day", "intervalday"); |
| allTypes.put("boolean", "bit"); |
| allTypes.put("varchar", "varchar"); |
| allTypes.put("time", "time"); |
| |
| List<String> allTypeSelectsAndCasts = new ArrayList<>(); |
| for (String s : allTypes.keySet()) { |
| // don't need to cast a varchar, just add the column reference |
| if (s.equals("varchar")) { |
| allTypeSelectsAndCasts.add(String.format("`%s_col`", allTypes.get(s))); |
| continue; |
| } |
| allTypeSelectsAndCasts.add(String.format("cast(`%s_col` AS %S) `%s_col`", allTypes.get(s), s, allTypes.get(s))); |
| } |
| allTypesSelection = Joiner.on(",").join(allTypeSelectsAndCasts); |
| } |
| |
| private final String allTypesTable = "cp.`parquet/alltypes.json`"; |
| |
| @Parameterized.Parameter |
| public int repeat = 1; |
| |
| @BeforeClass |
| public static void setUp() throws Exception { |
| startCluster(ClusterFixture.builder(dirTestWatcher)); |
| client.alterSession(PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY, true); |
| } |
| |
| @AfterClass |
| public static void disableDecimalDataType() { |
| client.resetSession(PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY); |
| } |
| |
| @Test |
| public void testSmallFileValueReadWrite() throws Exception { |
| String selection = "key"; |
| String inputTable = "cp.`store/json/intData.json`"; |
| runTestAndValidate(selection, selection, inputTable, "smallFileTest"); |
| } |
| |
| @Test |
| public void testSimple() throws Exception { |
| String selection = "*"; |
| String inputTable = "cp.`employee.json`"; |
| runTestAndValidate(selection, selection, inputTable, "employee_parquet"); |
| } |
| |
| @Test |
| public void testLargeFooter() throws Exception { |
| StringBuilder sb = new StringBuilder(); |
| // create a JSON document with a lot of columns |
| sb.append("{"); |
| final int numCols = 1000; |
| String[] colNames = new String[numCols]; |
| Object[] values = new Object[numCols]; |
| for (int i = 0; i < numCols - 1; i++) { |
| sb.append(String.format("\"col_%d\" : 100,", i)); |
| colNames[i] = "col_" + i; |
| values[i] = 100L; |
| } |
| // add one column without a comma after it |
| sb.append(String.format("\"col_%d\" : 100", numCols - 1)); |
| sb.append("}"); |
| colNames[numCols - 1] = "col_" + (numCols - 1); |
| values[numCols - 1] = 100L; |
| |
| String path = "test"; |
| File pathDir = dirTestWatcher.makeRootSubDir(Paths.get(path)); |
| |
| // write it to a file in the temp directory for the test |
| new TestConstantFolding.SmallFileCreator(pathDir) |
| .setRecord(sb.toString()).createFiles(1, 1, "json"); |
| |
| run("create table dfs.tmp.WIDE_PARQUET_TABLE_TestParquetWriter_testLargeFooter as select * from dfs.`%s/smallfile/smallfile.json`", path); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.WIDE_PARQUET_TABLE_TestParquetWriter_testLargeFooter") |
| .unOrdered() |
| .baselineColumns(colNames) |
| .baselineValues(values) |
| .build().run(); |
| } |
| |
| @Test |
| public void testAllScalarTypes() throws Exception { |
| /// read once with the flat reader |
| runTestAndValidate(allTypesSelection, "*", allTypesTable, "donuts_json"); |
| |
| try { |
| // read all of the types with the complex reader |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| runTestAndValidate(allTypesSelection, "*", allTypesTable, "donuts_json"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| @Test |
| public void testAllScalarTypesDictionary() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING, true); |
| /// read once with the flat reader |
| runTestAndValidate(allTypesSelection, "*", allTypesTable, "donuts_json"); |
| |
| // read all of the types with the complex reader |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| runTestAndValidate(allTypesSelection, "*", allTypesTable, "donuts_json"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| client.resetSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING); |
| } |
| } |
| |
| @Test |
| public void testDictionaryError() throws Exception { |
| compareParquetReadersColumnar("*", "cp.`parquet/required_dictionary.parquet`"); |
| runTestAndValidate("*", "*", "cp.`parquet/required_dictionary.parquet`", "required_dictionary"); |
| } |
| |
| @Test |
| public void testDictionaryEncoding() throws Exception { |
| String selection = "type"; |
| String inputTable = "cp.`donuts.json`"; |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING, true); |
| runTestAndValidate(selection, selection, inputTable, "donuts_json"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING); |
| } |
| } |
| |
| @Test |
| public void testComplex() throws Exception { |
| String selection = "*"; |
| String inputTable = "cp.`donuts.json`"; |
| runTestAndValidate(selection, selection, inputTable, "donuts_json"); |
| } |
| |
| @Test |
| public void testComplexRepeated() throws Exception { |
| String selection = "*"; |
| String inputTable = "cp.`testRepeatedWrite.json`"; |
| runTestAndValidate(selection, selection, inputTable, "repeated_json"); |
| } |
| |
| @Test |
| public void testCastProjectBug_Drill_929() throws Exception { |
| String selection = "L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER, L_QUANTITY, L_EXTENDEDPRICE, L_DISCOUNT, L_TAX, " + |
| "L_RETURNFLAG, L_LINESTATUS, L_SHIPDATE, cast(L_COMMITDATE as DATE) as COMMITDATE, cast(L_RECEIPTDATE as DATE) AS RECEIPTDATE, L_SHIPINSTRUCT, L_SHIPMODE, L_COMMENT"; |
| String validationSelection = "L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER, L_QUANTITY, L_EXTENDEDPRICE, L_DISCOUNT, L_TAX, " + |
| "L_RETURNFLAG, L_LINESTATUS, L_SHIPDATE,COMMITDATE ,RECEIPTDATE, L_SHIPINSTRUCT, L_SHIPMODE, L_COMMENT"; |
| |
| String inputTable = "cp.`tpch/lineitem.parquet`"; |
| runTestAndValidate(selection, validationSelection, inputTable, "drill_929"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite1() throws Exception { |
| String inputTable = "cp.`tpch/lineitem.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "lineitem_parquet_all"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite1_date_convertedType() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING, false); |
| String selection = "L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER, L_QUANTITY, L_EXTENDEDPRICE, L_DISCOUNT, L_TAX, " + |
| "L_RETURNFLAG, L_LINESTATUS, L_SHIPDATE, cast(L_COMMITDATE as DATE) as L_COMMITDATE, cast(L_RECEIPTDATE as DATE) AS L_RECEIPTDATE, L_SHIPINSTRUCT, L_SHIPMODE, L_COMMENT"; |
| String validationSelection = "L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER, L_QUANTITY, L_EXTENDEDPRICE, L_DISCOUNT, L_TAX, " + |
| "L_RETURNFLAG, L_LINESTATUS, L_SHIPDATE,L_COMMITDATE ,L_RECEIPTDATE, L_SHIPINSTRUCT, L_SHIPMODE, L_COMMENT"; |
| String inputTable = "cp.`tpch/lineitem.parquet`"; |
| runTestAndValidate(selection, validationSelection, inputTable, "lineitem_parquet_converted"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWrite2() throws Exception { |
| String inputTable = "cp.`tpch/customer.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "customer_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite3() throws Exception { |
| String inputTable = "cp.`tpch/nation.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "nation_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite4() throws Exception { |
| String inputTable = "cp.`tpch/orders.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "orders_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite5() throws Exception { |
| String inputTable = "cp.`tpch/part.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "part_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite6() throws Exception { |
| String inputTable = "cp.`tpch/partsupp.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "partsupp_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite7() throws Exception { |
| String inputTable = "cp.`tpch/region.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "region_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWrite8() throws Exception { |
| String inputTable = "cp.`tpch/supplier.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "supplier_parquet"); |
| } |
| |
| @Test |
| public void testTPCHReadWriteNoDictUncompressed() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING, false); |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "none"); |
| String inputTable = "cp.`tpch/supplier.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "supplier_parquet_no_dict_uncompressed"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_ENABLE_DICTIONARY_ENCODING); |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteDictGzip() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "gzip"); |
| String inputTable = "cp.`tpch/supplier.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "supplier_parquet_dict_gzip"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| } |
| } |
| |
| // working to create an exhaustive test of the format for this one. including all convertedTypes |
| // will not be supporting interval for Beta as of current schedule |
| // Types left out: |
| // "TIMESTAMPTZ_col" |
| @Test |
| public void testRepeated() throws Exception { |
| String inputTable = "cp.`parquet/basic_repeated.json`"; |
| runTestAndValidate("*", "*", inputTable, "basic_repeated"); |
| } |
| |
| @Test |
| public void testRepeatedDouble() throws Exception { |
| String inputTable = "cp.`parquet/repeated_double_data.json`"; |
| runTestAndValidate("*", "*", inputTable, "repeated_double_parquet"); |
| } |
| |
| @Test |
| public void testRepeatedLong() throws Exception { |
| String inputTable = "cp.`parquet/repeated_integer_data.json`"; |
| runTestAndValidate("*", "*", inputTable, "repeated_int_parquet"); |
| } |
| |
| @Test |
| public void testRepeatedBool() throws Exception { |
| String inputTable = "cp.`parquet/repeated_bool_data.json`"; |
| runTestAndValidate("*", "*", inputTable, "repeated_bool_parquet"); |
| } |
| |
| @Test |
| public void testNullReadWrite() throws Exception { |
| String inputTable = "cp.`parquet/null_test_data.json`"; |
| runTestAndValidate("*", "*", inputTable, "nullable_test"); |
| } |
| |
| @Ignore("Test file not available") |
| @Test |
| public void testBitError_Drill_2031() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/wide2/0_0_3.parquet`"); |
| } |
| |
| @Test |
| public void testDecimal() throws Exception { |
| String selection = "cast(salary as decimal(8,2)) as decimal8, cast(salary as decimal(15,2)) as decimal15, " + |
| "cast(salary as decimal(24,2)) as decimal24, cast(salary as decimal(38,2)) as decimal38"; |
| String validateSelection = "decimal8, decimal15, decimal24, decimal38"; |
| String inputTable = "cp.`employee.json`"; |
| |
| // DRILL-5833: The "old" writer had a decimal bug, but the new one |
| // did not. The one used was random. Force the test to run both |
| // the old and new readers. |
| |
| try { |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| runTestAndValidate(selection, validateSelection, inputTable, "parquet_decimal"); |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, false); |
| runTestAndValidate(selection, validateSelection, inputTable, "parquet_decimal"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| @Test |
| public void testMulipleRowGroups() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_BLOCK_SIZE, 1024*1024); |
| String selection = "mi"; |
| String inputTable = "cp.`customer.json`"; |
| runTestAndValidate(selection, selection, inputTable, "foodmart_customer_parquet"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_BLOCK_SIZE); |
| } |
| } |
| |
| @Test |
| public void testDate() throws Exception { |
| String selection = "cast(hire_date as DATE) as hire_date"; |
| String validateSelection = "hire_date"; |
| String inputTable = "cp.`employee.json`"; |
| runTestAndValidate(selection, validateSelection, inputTable, "foodmart_employee_parquet"); |
| } |
| |
| @Test |
| public void testBoolean() throws Exception { |
| String selection = "true as x, false as y"; |
| String validateSelection = "x, y"; |
| String inputTable = "cp.`tpch/region.parquet`"; |
| runTestAndValidate(selection, validateSelection, inputTable, "region_boolean_parquet"); |
| } |
| |
| @Test //DRILL-2030 |
| public void testWriterWithStarAndExp() throws Exception { |
| String selection = " *, r_regionkey + 1 r_regionkey2"; |
| String validateSelection = "r_regionkey, r_name, r_comment, r_regionkey + 1 r_regionkey2"; |
| String inputTable = "cp.`tpch/region.parquet`"; |
| runTestAndValidate(selection, validateSelection, inputTable, "region_star_exp"); |
| } |
| |
| @Test // DRILL-2458 |
| public void testWriterWithStarAndRegluarCol() throws Exception { |
| String outputFile = "region_sort"; |
| String ctasStmt = "create table " + outputFile + " as select *, r_regionkey + 1 as key1 from cp.`tpch/region.parquet` order by r_name"; |
| String query = "select r_regionkey, r_name, r_comment, r_regionkey +1 as key1 from cp.`tpch/region.parquet` order by r_name"; |
| String queryFromWriteOut = "select * from " + outputFile; |
| |
| try { |
| run("use dfs.tmp"); |
| run(ctasStmt); |
| testBuilder() |
| .ordered() |
| .sqlQuery(queryFromWriteOut) |
| .sqlBaselineQuery(query) |
| .build().run(); |
| } finally { |
| run("drop table if exists dfs.tmp.%s", outputFile); |
| } |
| } |
| |
| public void compareParquetReadersColumnar(String selection, String table) throws Exception { |
| String query = "select " + selection + " from " + table; |
| |
| try { |
| testBuilder() |
| .ordered() |
| .sqlQuery(query) |
| .optionSettingQueriesForTestQuery("alter session set `store.parquet.use_new_reader` = false") |
| .sqlBaselineQuery(query) |
| .optionSettingQueriesForBaseline("alter session set `store.parquet.use_new_reader` = true") |
| .build().run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| public void compareParquetReadersHyperVector(String selection, String table) throws Exception { |
| |
| String query = "select " + selection + " from " + table; |
| try { |
| testBuilder() |
| .ordered() |
| .highPerformanceComparison() |
| .sqlQuery(query) |
| .optionSettingQueriesForTestQuery( |
| "alter system set `store.parquet.use_new_reader` = false") |
| .sqlBaselineQuery(query) |
| .optionSettingQueriesForBaseline( |
| "alter system set `store.parquet.use_new_reader` = true") |
| .build().run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testReadVoter() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/voter.parquet`"); |
| } |
| |
| @Ignore("Test file not available") |
| @Test |
| public void testReadSf_100_supplier() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/sf100_supplier.parquet`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testParquetRead_checkNulls_NullsFirst() throws Exception { |
| compareParquetReadersColumnar("*", |
| "dfs.`tmp/parquet_with_nulls_should_sum_100000_nulls_first.parquet`"); |
| } |
| |
| @Ignore("Test file not available") |
| @Test |
| public void testParquetRead_checkNulls() throws Exception { |
| compareParquetReadersColumnar("*", "dfs.`tmp/parquet_with_nulls_should_sum_100000.parquet`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void test958_sql() throws Exception { |
| compareParquetReadersHyperVector("ss_ext_sales_price", "dfs.`tmp/store_sales`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testReadSf_1_supplier() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/orders_part-m-00001.parquet`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void test958_sql_all_columns() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/store_sales`"); |
| compareParquetReadersHyperVector("ss_addr_sk, ss_hdemo_sk", "dfs.`tmp/store_sales`"); |
| // TODO - Drill 1388 - this currently fails, but it is an issue with project, not the reader, pulled out the physical plan |
| // removed the unneeded project in the plan and ran it against both readers, they outputs matched |
| // compareParquetReadersHyperVector("pig_schema,ss_sold_date_sk,ss_item_sk,ss_cdemo_sk,ss_addr_sk, ss_hdemo_sk", |
| // "dfs.`tmp/store_sales`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testDrill_1314() throws Exception { |
| compareParquetReadersColumnar("l_partkey ", "dfs.`tmp/drill_1314.parquet`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testDrill_1314_all_columns() throws Exception { |
| compareParquetReadersHyperVector("*", "dfs.`tmp/drill_1314.parquet`"); |
| compareParquetReadersColumnar( |
| "l_orderkey,l_partkey,l_suppkey,l_linenumber, l_quantity, l_extendedprice,l_discount,l_tax", |
| "dfs.`tmp/drill_1314.parquet`"); |
| } |
| |
| @Ignore("Test file not available") |
| @Test |
| public void testParquetRead_checkShortNullLists() throws Exception { |
| compareParquetReadersColumnar("*", "dfs.`tmp/short_null_lists.parquet`"); |
| } |
| |
| @Ignore("Test file not available") |
| @Test |
| public void testParquetRead_checkStartWithNull() throws Exception { |
| compareParquetReadersColumnar("*", "dfs.`tmp/start_with_null.parquet`"); |
| } |
| |
| @Ignore("Binary file too large for version control") |
| @Test |
| public void testParquetReadWebReturns() throws Exception { |
| compareParquetReadersColumnar("wr_returning_customer_sk", "dfs.`tmp/web_returns`"); |
| } |
| |
| @Test |
| public void testWriteDecimal() throws Exception { |
| String outputTable = "decimal_test"; |
| |
| try { |
| run("create table dfs.tmp.%s as select " + |
| "cast('1.2' as decimal(38, 2)) col1, cast('1.2' as decimal(28, 2)) col2 " + |
| "from cp.`employee.json` limit 1", outputTable); |
| |
| BigDecimal result = new BigDecimal("1.20"); |
| |
| testBuilder() |
| .unOrdered() |
| .sqlQuery("select col1, col2 from dfs.tmp.%s ", outputTable) |
| .baselineColumns("col1", "col2") |
| .baselineValues(result, result) |
| .go(); |
| } finally { |
| run("drop table if exists dfs.tmp.%s", outputTable); |
| } |
| } |
| |
| @Test // DRILL-2341 |
| @Category(UnlikelyTest.class) |
| public void tableSchemaWhenSelectFieldsInDef_SelectFieldsInView() throws Exception { |
| final String newTblName = "testTableOutputSchema"; |
| |
| try { |
| run("CREATE TABLE dfs.tmp.%s(id, name, bday) AS SELECT " + |
| "cast(`employee_id` as integer), " + |
| "cast(`full_name` as varchar(100)), " + |
| "cast(`birth_date` as date) " + |
| "FROM cp.`employee.json` ORDER BY `employee_id` LIMIT 1", newTblName); |
| |
| testBuilder() |
| .unOrdered() |
| .sqlQuery("SELECT * FROM dfs.tmp.`%s`", newTblName) |
| .baselineColumns("id", "name", "bday") |
| .baselineValues(1, "Sheri Nowmer", LocalDate.parse("1961-08-26")) |
| .go(); |
| } finally { |
| run("drop table if exists dfs.tmp.%s", newTblName); |
| } |
| } |
| |
| /* |
| * Method tests CTAS with interval data type. We also verify reading back the data to ensure we |
| * have written the correct type. For every CTAS operation we use both the readers to verify results. |
| */ |
| @Test |
| public void testCTASWithIntervalTypes() throws Exception { // TODO: investigate NPE errors during the test execution |
| run("use dfs.tmp"); |
| |
| String tableName = "drill_1980_t1"; |
| // test required interval day type |
| run("create table %s as " + |
| "select " + |
| "interval '10 20:30:40.123' day to second col1, " + |
| "interval '-1000000000 20:12:23.999' day(10) to second col2 " + |
| "from cp.`employee.json` limit 2", tableName); |
| |
| Period row1Col1 = new Period(0, 0, 0, 10, 0, 0, 0, 73840123); |
| Period row1Col2 = new Period(0, 0, 0, -1000000000, 0, 0, 0, -72743999); |
| testParquetReaderHelper(tableName, row1Col1, row1Col2, row1Col1, row1Col2); |
| |
| tableName = "drill_1980_2"; |
| |
| // test required interval year type |
| run("create table %s as " + |
| "select " + |
| "interval '10-2' year to month col1, " + |
| "interval '-100-8' year(3) to month col2 " + |
| "from cp.`employee.json` limit 2", tableName); |
| |
| row1Col1 = new Period(0, 122, 0, 0, 0, 0, 0, 0); |
| row1Col2 = new Period(0, -1208, 0, 0, 0, 0, 0, 0); |
| |
| testParquetReaderHelper(tableName, row1Col1, row1Col2, row1Col1, row1Col2); |
| // test nullable interval year type |
| tableName = "drill_1980_t3"; |
| run("create table %s as " + |
| "select " + |
| "cast (intervalyear_col as interval year) col1," + |
| "cast(intervalyear_col as interval year) + interval '2' year col2 " + |
| "from cp.`parquet/alltypes.json` where tinyint_col = 1 or tinyint_col = 2", tableName); |
| |
| row1Col1 = new Period(0, 12, 0, 0, 0, 0, 0, 0); |
| row1Col2 = new Period(0, 36, 0, 0, 0, 0, 0, 0); |
| Period row2Col1 = new Period(0, 24, 0, 0, 0, 0, 0, 0); |
| Period row2Col2 = new Period(0, 48, 0, 0, 0, 0, 0, 0); |
| |
| testParquetReaderHelper(tableName, row1Col1, row1Col2, row2Col1, row2Col2); |
| |
| // test nullable interval day type |
| tableName = "drill_1980_t4"; |
| run("create table %s as " + |
| "select " + |
| "cast(intervalday_col as interval day) col1, " + |
| "cast(intervalday_col as interval day) + interval '1' day col2 " + |
| "from cp.`parquet/alltypes.json` where tinyint_col = 1 or tinyint_col = 2", tableName); |
| |
| row1Col1 = new Period(0, 0, 0, 1, 0, 0, 0, 0); |
| row1Col2 = new Period(0, 0, 0, 2, 0, 0, 0, 0); |
| row2Col1 = new Period(0, 0, 0, 2, 0, 0, 0, 0); |
| row2Col2 = new Period(0, 0, 0, 3, 0, 0, 0, 0); |
| |
| testParquetReaderHelper(tableName, row1Col1, row1Col2, row2Col1, row2Col2); |
| } |
| |
| private void testParquetReaderHelper(String tableName, Period row1Col1, Period row1Col2, |
| Period row2Col1, Period row2Col2) throws Exception { |
| |
| final String switchReader = "alter session set `store.parquet.use_new_reader` = %s; "; |
| final String enableVectorizedReader = String.format(switchReader, true); |
| final String disableVectorizedReader = String.format(switchReader, false); |
| String query = String.format("select * from %s", tableName); |
| |
| testBuilder() |
| .sqlQuery(query) |
| .unOrdered() |
| .optionSettingQueriesForTestQuery(enableVectorizedReader) |
| .baselineColumns("col1", "col2") |
| .baselineValues(row1Col1, row1Col2) |
| .baselineValues(row2Col1, row2Col2) |
| .go(); |
| |
| testBuilder() |
| .sqlQuery(query) |
| .unOrdered() |
| .optionSettingQueriesForTestQuery(disableVectorizedReader) |
| .baselineColumns("col1", "col2") |
| .baselineValues(row1Col1, row1Col2) |
| .baselineValues(row2Col1, row2Col2) |
| .go(); |
| } |
| |
| public void runTestAndValidate(String selection, String validationSelection, String inputTable, String outputFile) throws Exception { |
| try { |
| run("drop table if exists dfs.tmp.%s", outputFile); |
| |
| final String query = String.format("SELECT %s FROM %s", selection, inputTable); |
| |
| run("use dfs.tmp"); |
| run("CREATE TABLE %s AS %s", outputFile, query); |
| testBuilder() |
| .unOrdered() |
| .sqlQuery(query) |
| .sqlBaselineQuery("SELECT %s FROM %s", validationSelection, outputFile) |
| .go(); |
| |
| Configuration hadoopConf = new Configuration(); |
| hadoopConf.set(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS); |
| |
| Path output = new Path(dirTestWatcher.getDfsTestTmpDir().getAbsolutePath(), outputFile); |
| FileSystem fs = output.getFileSystem(hadoopConf); |
| for (FileStatus file : fs.listStatus(output)) { |
| ParquetMetadata footer = ParquetFileReader.readFooter(hadoopConf, file, SKIP_ROW_GROUPS); |
| String version = footer.getFileMetaData().getKeyValueMetaData().get(DRILL_VERSION_PROPERTY); |
| assertEquals(DrillVersionInfo.getVersion(), version); |
| } |
| } finally { |
| run("drop table if exists dfs.tmp.%s", outputFile); |
| } |
| } |
| |
| /* |
| Impala encodes timestamp values as int96 fields. Test the reading of an int96 field with two converters: |
| the first one converts parquet INT96 into drill VARBINARY and the second one (works while |
| store.parquet.reader.int96_as_timestamp option is enabled) converts parquet INT96 into drill TIMESTAMP. |
| */ |
| @Test |
| public void testImpalaParquetInt96() throws Exception { |
| compareParquetReadersColumnar("field_impala_ts", "cp.`parquet/int96_impala_1.parquet`"); |
| try { |
| client.alterSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP, true); |
| compareParquetReadersColumnar("field_impala_ts", "cp.`parquet/int96_impala_1.parquet`"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP); |
| } |
| } |
| |
| /* |
| Test the reading of a binary field as drill varbinary where data is in dictionary _and_ non-dictionary encoded pages |
| */ |
| @Test |
| public void testImpalaParquetBinaryAsVarBinary_DictChange() throws Exception { |
| compareParquetReadersColumnar("field_impala_ts", "cp.`parquet/int96_dict_change.parquet`"); |
| } |
| |
| /* |
| Test the reading of a binary field as drill timestamp where data is in dictionary _and_ non-dictionary encoded pages |
| */ |
| @Test |
| public void testImpalaParquetBinaryAsTimeStamp_DictChange() throws Exception { |
| try { |
| testBuilder() |
| .sqlQuery("select min(int96_ts) date_value from dfs.`parquet/int96_dict_change`") |
| .optionSettingQueriesForTestQuery( |
| "alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP) |
| .ordered() |
| .baselineColumns("date_value") |
| .baselineValues(convertToLocalDateTime("1970-01-01 00:00:01.000")) |
| .build().run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP); |
| } |
| } |
| |
| @Test |
| public void testSparkParquetBinaryAsTimeStamp_DictChange() throws Exception { |
| try { |
| testBuilder() |
| .sqlQuery("select distinct run_date from cp.`parquet/spark-generated-int96-timestamp.snappy.parquet`") |
| .optionSettingQueriesForTestQuery( |
| "alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP) |
| .ordered() |
| .baselineColumns("run_date") |
| .baselineValues(convertToLocalDateTime("2017-12-06 16:38:43.988")) |
| .build().run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP); |
| } |
| } |
| |
| /* |
| Test the conversion from int96 to impala timestamp |
| */ |
| @Test |
| public void testTimestampImpalaConvertFrom() throws Exception { |
| compareParquetReadersColumnar("convert_from(field_impala_ts, 'TIMESTAMP_IMPALA')", "cp.`parquet/int96_impala_1.parquet`"); |
| } |
| |
| /* |
| Test reading parquet Int96 as TimeStamp and comparing obtained values with the |
| old results (reading the same values as VarBinary and convert_fromTIMESTAMP_IMPALA function using) |
| */ |
| @Test |
| public void testImpalaParquetTimestampInt96AsTimeStamp() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, false); |
| compareParquetInt96Converters("field_impala_ts", "cp.`parquet/int96_impala_1.parquet`"); |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| compareParquetInt96Converters("field_impala_ts", "cp.`parquet/int96_impala_1.parquet`"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| /* |
| Test a file with partitions and an int96 column. (Data generated using Hive) |
| */ |
| @Test |
| public void testImpalaParquetInt96Partitioned() throws Exception { |
| compareParquetReadersColumnar("timestamp_field", "cp.`parquet/part1/hive_all_types.parquet`"); |
| } |
| |
| /* |
| Test the conversion from int96 to impala timestamp with hive data including nulls. Validate against old reader |
| */ |
| @Test |
| public void testHiveParquetTimestampAsInt96_compare() throws Exception { |
| compareParquetReadersColumnar("convert_from(timestamp_field, 'TIMESTAMP_IMPALA')", |
| "cp.`parquet/part1/hive_all_types.parquet`"); |
| } |
| |
| /* |
| Test the conversion from int96 to impala timestamp with hive data including nulls. Validate against expected values |
| */ |
| @Test |
| public void testHiveParquetTimestampAsInt96_basic() throws Exception { |
| testBuilder() |
| .unOrdered() |
| .sqlQuery("SELECT convert_from(timestamp_field, 'TIMESTAMP_IMPALA') as timestamp_field " |
| + "from cp.`parquet/part1/hive_all_types.parquet` ") |
| .baselineColumns("timestamp_field") |
| .baselineValues(convertToLocalDateTime("2013-07-06 00:01:00")) |
| .baselineValues((Object)null) |
| .go(); |
| } |
| |
| @Test |
| @Ignore |
| public void testSchemaChange() throws Exception { |
| File dir = new File("target/" + this.getClass()); |
| if ((!dir.exists() && !dir.mkdirs()) || (dir.exists() && !dir.isDirectory())) { |
| throw new RuntimeException("can't create dir " + dir); |
| } |
| File input1 = new File(dir, "1.json"); |
| File input2 = new File(dir, "2.json"); |
| try (FileWriter fw = new FileWriter(input1)) { |
| fw.append("{\"a\":\"foo\"}\n"); |
| } |
| try (FileWriter fw = new FileWriter(input2)) { |
| fw.append("{\"b\":\"foo\"}\n"); |
| } |
| run("select * from " + "dfs.`" + dir.getAbsolutePath() + "`"); |
| runTestAndValidate("*", "*", "dfs.`" + dir.getAbsolutePath() + "`", "schema_change_parquet"); |
| } |
| |
| |
| /* |
| The following test boundary conditions for null values occurring on page boundaries. All files have at least one dictionary |
| encoded page for all columns |
| */ |
| @Test |
| public void testAllNulls() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/all_nulls.parquet`"); |
| } |
| |
| @Test |
| public void testNoNulls() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/no_nulls.parquet`"); |
| } |
| |
| @Test |
| public void testFirstPageAllNulls() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/first_page_all_nulls.parquet`"); |
| } |
| @Test |
| public void testLastPageAllNulls() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/last_page_all_nulls.parquet`"); |
| } |
| @Test |
| public void testFirstPageOneNull() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/first_page_one_null.parquet`"); |
| } |
| @Test |
| public void testLastPageOneNull() throws Exception { |
| compareParquetReadersColumnar( |
| "c_varchar, c_integer, c_bigint, c_float, c_double, c_date, c_time, c_timestamp, c_boolean", |
| "cp.`parquet/last_page_one_null.parquet`"); |
| } |
| |
| private void compareParquetInt96Converters(String selection, String table) throws Exception { |
| try { |
| testBuilder() |
| .ordered() |
| .sqlQuery("select `%1$s` from %2$s order by `%1$s`", selection, table) |
| .optionSettingQueriesForTestQuery( |
| "alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP) |
| .sqlBaselineQuery("select convert_from(`%1$s`, 'TIMESTAMP_IMPALA') as `%1$s` from %2$s order by `%1$s`", |
| selection, table) |
| .optionSettingQueriesForBaseline( |
| "alter session set `%s` = false", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP) |
| .build() |
| .run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteFormatV2() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_FORMAT_VERSION, "v2"); |
| String inputTable = "cp.`tpch/supplier.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_v2"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_FORMAT_VERSION); |
| } |
| } |
| |
| @Ignore ("Used to test decompression in AsyncPageReader. Takes too long.") |
| @Test |
| public void testTPCHReadWriteRunRepeated() throws Exception { |
| for (int i = 1; i <= repeat; i++) { |
| testTPCHReadWriteGzip(); |
| testTPCHReadWriteSnappy(); |
| testTPCHReadWriteBrotli(); |
| testTPCHReadWriteLz4(); |
| testTPCHReadWriteLzo(); |
| testTPCHReadWriteZstd(); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteSnappy() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "snappy"); |
| String inputTable = "cp.`tpch/supplier.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_snappy"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteGzip() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "gzip"); |
| String inputTable = "cp.`supplier_gzip.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_gzip"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| } |
| } |
| |
| // Only attempt this test on Linux / amd64 because com.rdblue.brotli-codec |
| // only bundles natives for Mac and Linux on AMD64. See PARQUET-1975. |
| @Test |
| public void testTPCHReadWriteBrotli() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "brotli"); |
| // exercise the new Parquet record reader with this parquet-mr-backed codec |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| String inputTable = "cp.`supplier_brotli.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_brotli"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteLz4() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "lz4"); |
| // TODO: Uncomment the below to reveal DRILL-8138. It is commented out now |
| // to allow a clean test run for the release of 1.20, since this bug is not |
| // a blocker. |
| // Exercise the async Parquet column reader with this aircompressor-backed codec |
| // client.alterSession(ExecConstants.PARQUET_COLUMNREADER_ASYNC, true); |
| client.alterSession(ExecConstants.PARQUET_PAGEREADER_ASYNC, false); |
| String inputTable = "cp.`supplier_lz4.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_lz4"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| // client.resetSession(ExecConstants.PARQUET_COLUMNREADER_ASYNC); |
| client.resetSession(ExecConstants.PARQUET_PAGEREADER_ASYNC); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteLzo() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "lzo"); |
| // exercise the async Parquet page reader with this aircompressor-backed codec |
| client.alterSession(ExecConstants.PARQUET_PAGEREADER_ASYNC, true); |
| String inputTable = "cp.`supplier_lzo.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_lzo"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| client.resetSession(ExecConstants.PARQUET_PAGEREADER_ASYNC); |
| } |
| } |
| |
| @Test |
| public void testTPCHReadWriteZstd() throws Exception { |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE, "zstd"); |
| // exercise the new Parquet record reader with this aircompressor-backed codec |
| client.alterSession(ExecConstants.PARQUET_NEW_RECORD_READER, true); |
| String inputTable = "cp.`supplier_zstd.parquet`"; |
| runTestAndValidate("*", "*", inputTable, "suppkey_parquet_dict_zstd"); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_COMPRESSION_TYPE); |
| client.resetSession(ExecConstants.PARQUET_NEW_RECORD_READER); |
| } |
| } |
| |
| @Test // DRILL-5097 |
| @Category(UnlikelyTest.class) |
| public void testInt96TimeStampValueWidth() throws Exception { |
| try { |
| testBuilder() |
| .unOrdered() |
| .sqlQuery("select c, d from cp.`parquet/data.snappy.parquet` " + |
| "where `a` is not null and `c` is not null and `d` is not null") |
| .optionSettingQueriesForTestQuery( |
| "alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP) |
| .baselineColumns("c", "d") |
| .baselineValues(LocalDate.parse("2012-12-15"), |
| convertToLocalDateTime("2016-04-24 20:06:28")) |
| .baselineValues(LocalDate.parse("2011-07-09"), |
| convertToLocalDateTime("2015-04-15 22:35:49")) |
| .build() |
| .run(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalIntBigIntFixedLen() throws Exception { |
| String tableName = "decimalIntBigIntFixedLen"; |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, FIXED_LEN_BYTE_ARRAY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select cast('123456.789' as decimal(9, 3)) as decInt,\n" + |
| "cast('123456.789123456789' as decimal(18, 12)) as decBigInt,\n" + |
| "cast('123456.789123456789' as decimal(19, 12)) as fixedLen", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decInt", INT32), |
| Pair.of("decBigInt", INT64), |
| Pair.of("fixedLen", FIXED_LEN_BYTE_ARRAY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decInt", "decBigInt", "fixedLen") |
| .baselineValues(new BigDecimal("123456.789"), |
| new BigDecimal("123456.789123456789"), |
| new BigDecimal("123456.789123456789")) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalIntBigIntBinary() throws Exception { |
| String tableName = "decimalIntBigIntBinary"; |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, true); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, BINARY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select cast('123456.789' as decimal(9, 3)) as decInt,\n" + |
| "cast('123456.789123456789' as decimal(18, 12)) as decBigInt,\n" + |
| "cast('123456.789123456789' as decimal(19, 12)) as binCol", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decInt", INT32), |
| Pair.of("decBigInt", INT64), |
| Pair.of("binCol", BINARY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decInt", "decBigInt", "binCol") |
| .baselineValues(new BigDecimal("123456.789"), |
| new BigDecimal("123456.789123456789"), |
| new BigDecimal("123456.789123456789")) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalFixedLenOnly() throws Exception { |
| String tableName = "decimalFixedLenOnly"; |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, false); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, FIXED_LEN_BYTE_ARRAY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select cast('123456.789' as decimal(9, 3)) as decInt,\n" + |
| "cast('123456.789123456789' as decimal(18, 12)) as decBigInt,\n" + |
| "cast('123456.789123456789' as decimal(19, 12)) as fixedLen", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decInt", FIXED_LEN_BYTE_ARRAY), |
| Pair.of("decBigInt", FIXED_LEN_BYTE_ARRAY), |
| Pair.of("fixedLen", FIXED_LEN_BYTE_ARRAY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decInt", "decBigInt", "fixedLen") |
| .baselineValues(new BigDecimal("123456.789"), |
| new BigDecimal("123456.789123456789"), |
| new BigDecimal("123456.789123456789")) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalBinaryOnly() throws Exception { |
| String tableName = "decimalBinaryOnly"; |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, false); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, BINARY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select cast('123456.789' as decimal(9, 3)) as decInt,\n" + |
| "cast('123456.789123456789' as decimal(18, 12)) as decBigInt,\n" + |
| "cast('123456.789123456789' as decimal(19, 12)) as binCol", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decInt", BINARY), |
| Pair.of("decBigInt", BINARY), |
| Pair.of("binCol", BINARY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decInt", "decBigInt", "binCol") |
| .baselineValues(new BigDecimal("123456.789"), |
| new BigDecimal("123456.789123456789"), |
| new BigDecimal("123456.789123456789")) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalIntBigIntRepeated() throws Exception { |
| String tableName = "decimalIntBigIntRepeated"; |
| |
| JsonStringArrayList<BigDecimal> ints = new JsonStringArrayList<>(); |
| ints.add(new BigDecimal("999999.999")); |
| ints.add(new BigDecimal("-999999.999")); |
| ints.add(new BigDecimal("0.000")); |
| |
| JsonStringArrayList<BigDecimal> longs = new JsonStringArrayList<>(); |
| longs.add(new BigDecimal("999999999.999999999")); |
| longs.add(new BigDecimal("-999999999.999999999")); |
| longs.add(new BigDecimal("0.000000000")); |
| |
| JsonStringArrayList<BigDecimal> fixedLen = new JsonStringArrayList<>(); |
| fixedLen.add(new BigDecimal("999999999999.999999")); |
| fixedLen.add(new BigDecimal("-999999999999.999999")); |
| fixedLen.add(new BigDecimal("0.000000")); |
| |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, true); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, FIXED_LEN_BYTE_ARRAY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select * from cp.`parquet/repeatedIntLondFixedLenBinaryDecimal.parquet`", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decimal_int32", INT32), |
| Pair.of("decimal_int64", INT64), |
| Pair.of("decimal_fixedLen", INT64), |
| Pair.of("decimal_binary", INT64)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decimal_int32", "decimal_int64", "decimal_fixedLen", "decimal_binary") |
| .baselineValues(ints, longs, fixedLen, fixedLen) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalFixedLenRepeated() throws Exception { |
| String tableName = "decimalFixedLenRepeated"; |
| |
| JsonStringArrayList<BigDecimal> ints = new JsonStringArrayList<>(); |
| ints.add(new BigDecimal("999999.999")); |
| ints.add(new BigDecimal("-999999.999")); |
| ints.add(new BigDecimal("0.000")); |
| |
| JsonStringArrayList<BigDecimal> longs = new JsonStringArrayList<>(); |
| longs.add(new BigDecimal("999999999.999999999")); |
| longs.add(new BigDecimal("-999999999.999999999")); |
| longs.add(new BigDecimal("0.000000000")); |
| |
| JsonStringArrayList<BigDecimal> fixedLen = new JsonStringArrayList<>(); |
| fixedLen.add(new BigDecimal("999999999999.999999")); |
| fixedLen.add(new BigDecimal("-999999999999.999999")); |
| fixedLen.add(new BigDecimal("0.000000")); |
| |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, false); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, FIXED_LEN_BYTE_ARRAY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select * from cp.`parquet/repeatedIntLondFixedLenBinaryDecimal.parquet`", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decimal_int32", FIXED_LEN_BYTE_ARRAY), |
| Pair.of("decimal_int64", FIXED_LEN_BYTE_ARRAY), |
| Pair.of("decimal_fixedLen", FIXED_LEN_BYTE_ARRAY), |
| Pair.of("decimal_binary", FIXED_LEN_BYTE_ARRAY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decimal_int32", "decimal_int64", "decimal_fixedLen", "decimal_binary") |
| .baselineValues(ints, longs, fixedLen, fixedLen) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testWriteDecimalBinaryRepeated() throws Exception { |
| String tableName = "decimalBinaryRepeated"; |
| |
| JsonStringArrayList<BigDecimal> ints = new JsonStringArrayList<>(); |
| ints.add(new BigDecimal("999999.999")); |
| ints.add(new BigDecimal("-999999.999")); |
| ints.add(new BigDecimal("0.000")); |
| |
| JsonStringArrayList<BigDecimal> longs = new JsonStringArrayList<>(); |
| longs.add(new BigDecimal("999999999.999999999")); |
| longs.add(new BigDecimal("-999999999.999999999")); |
| longs.add(new BigDecimal("0.000000000")); |
| |
| JsonStringArrayList<BigDecimal> fixedLen = new JsonStringArrayList<>(); |
| fixedLen.add(new BigDecimal("999999999999.999999")); |
| fixedLen.add(new BigDecimal("-999999999999.999999")); |
| fixedLen.add(new BigDecimal("0.000000")); |
| try { |
| client.alterSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS, false); |
| client.alterSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS, BINARY.name()); |
| run( |
| "create table dfs.tmp.%s as\n" + |
| "select * from cp.`parquet/repeatedIntLondFixedLenBinaryDecimal.parquet`", tableName); |
| checkTableTypes(tableName, |
| ImmutableList.of( |
| Pair.of("decimal_int32", BINARY), |
| Pair.of("decimal_int64", BINARY), |
| Pair.of("decimal_fixedLen", BINARY), |
| Pair.of("decimal_binary", BINARY)), |
| true); |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.%s", tableName) |
| .unOrdered() |
| .baselineColumns("decimal_int32", "decimal_int64", "decimal_fixedLen", "decimal_binary") |
| .baselineValues(ints, longs, fixedLen, fixedLen) |
| .go(); |
| } finally { |
| client.resetSession(ExecConstants.PARQUET_WRITER_USE_PRIMITIVE_TYPES_FOR_DECIMALS); |
| client.resetSession(ExecConstants.PARQUET_WRITER_LOGICAL_TYPE_FOR_DECIMALS); |
| run("drop table if exists dfs.tmp.%s", tableName); |
| } |
| } |
| |
| @Test |
| public void testCtasForList() throws Exception { |
| String tableName = "testCtasForList"; |
| try { |
| run("CREATE TABLE `%s`.`%s` AS SELECT l FROM cp.`jsoninput/input2.json`", DFS_TMP_SCHEMA, tableName); |
| testBuilder() |
| .sqlQuery("SELECT * FROM `%s`.`/%s` LIMIT 1", DFS_TMP_SCHEMA, tableName) |
| .unOrdered() |
| .baselineColumns("l") |
| .baselineValues(asList(4L, 2L)) |
| .go(); |
| } finally { |
| run("DROP TABLE IF EXISTS `%s`.`%s`", DFS_TMP_SCHEMA, tableName); |
| } |
| } |
| |
| @Test |
| public void testCtasForRepeatedList() throws Exception { |
| String tableName = "testCtasForRepeatedList"; |
| try { |
| run("CREATE TABLE `%s`.`%s` AS SELECT * FROM cp.`jsoninput/repeated_list_bug.json`", DFS_TMP_SCHEMA, tableName); |
| testBuilder() |
| .sqlQuery("SELECT rl FROM `%s`.`/%s`", DFS_TMP_SCHEMA, tableName) |
| .unOrdered() |
| .baselineColumns("rl") |
| .baselineValues(asList(asList(4L, 6L), asList(2L, 3L))) |
| .baselineValues(asList(asList(9L, 7L), asList(4L, 8L))) |
| .go(); |
| } finally { |
| run("DROP TABLE IF EXISTS `%s`.`%s`", DFS_TMP_SCHEMA, tableName); |
| } |
| } |
| |
| @Test |
| public void testCtasForRepeatedListOfMaps() throws Exception { |
| String tableName = "testCtasForRepeatedListOfMaps"; |
| try { |
| run("CREATE TABLE `%s`.`%s` AS SELECT * FROM cp.`jsoninput/repeated_list_of_maps.json`", DFS_TMP_SCHEMA, tableName); |
| testBuilder() |
| .sqlQuery("SELECT * FROM `%s`.`/%s`", DFS_TMP_SCHEMA, tableName) |
| .unOrdered() |
| .baselineColumns("rma") |
| .baselineValues(asList( |
| asList(mapOf("a", 1L, "b", "2"), mapOf("a", 2L, "b", "3")), |
| asList(mapOf("a", 3L, "b", "3")) |
| )) |
| .baselineValues(asList( |
| asList(mapOf("a", 4L, "b", "4"), mapOf("a", 5L, "b", "5"), mapOf("a", 6L, "b", "6")), |
| asList(mapOf("a", 7L, "b", "7")) |
| )) |
| .go(); |
| } finally { |
| run("DROP TABLE IF EXISTS `%s`.`%s`", DFS_TMP_SCHEMA, tableName); |
| } |
| } |
| |
| @Test |
| public void testCTASWithDictInSelect() throws Exception { |
| String tableName = "table_with_dict"; |
| try { |
| run("use dfs.tmp"); |
| run("create table %s as select id, mapcol from cp.`store/parquet/complex/map/parquet/000000_0.parquet`", tableName); |
| testBuilder() |
| .sqlQuery("select * from %s", tableName) |
| .unOrdered() |
| .baselineColumns("id", "mapcol") |
| .baselineValues(1, TestBuilder.mapOfObject("b", 6, "c", 7)) |
| .baselineValues(3, TestBuilder.mapOfObject("b", null, "c", 8, "d", 9, "e", 10)) |
| .baselineValues(5, TestBuilder.mapOfObject("b", 6, "c", 7, "a", 8, "abc4", 9, "bde", 10)) |
| .baselineValues(4, TestBuilder.mapOfObject("a", 3, "b", 4, "c", 5)) |
| .baselineValues(2, TestBuilder.mapOfObject("a", 1, "b", 2, "c", 3)) |
| .go(); |
| } finally { |
| run("DROP TABLE IF EXISTS %s", tableName); |
| } |
| } |
| |
| @Test |
| public void testCTASWithRepeatedDictInSelect() throws Exception { |
| String tableName = "table_with_dict_array"; |
| try { |
| run("use dfs.tmp"); |
| run("create table %s as select id, map_array from cp.`store/parquet/complex/map/parquet/000000_0.parquet`", tableName); |
| testBuilder() |
| .sqlQuery("select * from %s", tableName) |
| .unOrdered() |
| .baselineColumns("id", "map_array") |
| .baselineValues( |
| 4, |
| TestBuilder.listOf( |
| TestBuilder.mapOfObject(1L, 2, 10L, 1, 42L, 3, 31L, 4), |
| TestBuilder.mapOfObject(-1L, 2, 3L, 1, 5L, 3, 54L, 4, 55L, 589, -78L, 2), |
| TestBuilder.mapOfObject(1L, 124, 3L, 1, -4L, 2, 19L, 3, 5L, 3, 9L, 1), |
| TestBuilder.mapOfObject(1L, 89, 2L, 1, 3L, 3, 4L, 21, 5L, 12, 6L, 34), |
| TestBuilder.mapOfObject(1L, -25, 3L, 1, 5L, 3, 6L, 2, 9L, 333, 10L, 344), |
| TestBuilder.mapOfObject(3L, 222, 4L, 1, 5L, 3, 6L, 2, 7L, 1, 8L, 3), |
| TestBuilder.mapOfObject(1L, 11, 3L, 12, 5L, 13) |
| ) |
| ) |
| .baselineValues( |
| 1, |
| TestBuilder.listOf( |
| TestBuilder.mapOfObject(8L, 1, 9L, 2, 523L, 4, 31L, 3), |
| TestBuilder.mapOfObject(1L, 2, 3L, 1, 5L, 3) |
| ) |
| ) |
| .baselineValues( |
| 3, |
| TestBuilder.listOf( |
| TestBuilder.mapOfObject(3L, 1), |
| TestBuilder.mapOfObject(1L, 2) |
| ) |
| ) |
| .baselineValues( |
| 2, |
| TestBuilder.listOf( |
| TestBuilder.mapOfObject(1L, 1, 2L, 2) |
| ) |
| ) |
| .baselineValues( |
| 5, |
| TestBuilder.listOf( |
| TestBuilder.mapOfObject(1L, 1, 2L, 2, 3L, 3, 4L, 4), |
| TestBuilder.mapOfObject(1L, -1, 2L, -2), |
| TestBuilder.mapOfObject(1L, 4, 2L, 5, 3L, 7) |
| ) |
| ) |
| .go(); |
| } finally { |
| run("DROP TABLE IF EXISTS %s", tableName); |
| } |
| } |
| |
| @Test |
| public void testFormatConfigOpts() throws Exception { |
| FileSystemConfig pluginConfig = (FileSystemConfig) cluster.storageRegistry().copyConfig("dfs"); |
| FormatPluginConfig backupConfig = pluginConfig.getFormats().get("parquet"); |
| |
| cluster.defineFormat("dfs", "parquet", new ParquetFormatConfig( |
| false, |
| true, |
| 123, |
| 456, |
| true, |
| "snappy", |
| "binary", |
| true, |
| "v2" |
| ) |
| ); |
| |
| try { |
| String query = "select * from dfs.`parquet/int96_dict_change`"; |
| queryBuilder() |
| .sql(query) |
| .jsonPlanMatcher() |
| .include("\"autoCorrectCorruptDates\" : false") |
| .include("\"enableStringsSignedMinMax\" : true") |
| .include("\"blockSize\" : 123") |
| .include("\"pageSize\" : 456") |
| .include("\"useSingleFSBlock\" : true") |
| .include("\"writerCompressionType\" : \"snappy\"") |
| .include("\"writerUsePrimitivesForDecimals\" : true") |
| .include("\"writerFormatVersion\" : \"v2\"") |
| .match(); |
| } finally { |
| cluster.defineFormat("dfs", "parquet", backupConfig); |
| } |
| } |
| |
| @Test // DRILL-8272 |
| public void testResultWithEmptyMap() throws Exception { |
| String fileName = "emptyMap.json"; |
| |
| // Create a test JSON object that includes two nested empty objects. The |
| // first has a string property as sibling so its parent, non_empty_child |
| // should not be eliminated from the output schema while the second has no |
| // siblings so its parent empty_child should be eliminated. |
| FileUtils.writeStringToFile( |
| new File(dirTestWatcher.getRootDir(), fileName), |
| "{\"non_empty_child\": { \"empty\": {}, \"b\": \"b\"}, \"empty_child\": { \"empty\": {} }, \"a\": \"a\"}", |
| Charset.defaultCharset() |
| ); |
| |
| run("create table dfs.tmp.t1 as SELECT * from dfs.`%s` t", fileName); |
| |
| testBuilder() |
| .sqlQuery("select * from dfs.tmp.t1") |
| .unOrdered() |
| .baselineColumns("a", "non_empty_child") |
| .baselineValues("a", mapOf("b", "b")) |
| .go(); |
| } |
| |
| /** |
| * Checks that specified parquet table contains specified columns with specified types. |
| * |
| * @param tableName name of the table that should be checked. |
| * @param columnsToCheck pair of column name and column type that should be checked in the table. |
| * @param isDecimalType is should be specified columns annotated ad DECIMAL. |
| * @throws IOException If table file was not found. |
| */ |
| private void checkTableTypes(String tableName, |
| List<Pair<String, PrimitiveType.PrimitiveTypeName>> columnsToCheck, |
| boolean isDecimalType) throws IOException { |
| MessageType schema = ParquetFileReader.readFooter( |
| new Configuration(), |
| new Path(Paths.get(dirTestWatcher.getDfsTestTmpDir().getPath(), tableName, "0_0_0.parquet").toUri().getPath()), |
| NO_FILTER).getFileMetaData().getSchema(); |
| |
| for (Pair<String, PrimitiveType.PrimitiveTypeName> nameType : columnsToCheck) { |
| assertEquals( |
| nameType.getValue(), |
| schema.getType(nameType.getKey()).asPrimitiveType().getPrimitiveTypeName(), |
| String.format( |
| "Table %s does not contain column %s with type %s", |
| tableName, |
| nameType.getKey(), |
| nameType.getValue() |
| ) |
| ); |
| |
| assertEquals( |
| isDecimalType, |
| schema.getType(nameType.getKey()).getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation, |
| String.format( |
| "Table %s %s column %s with DECIMAL type", |
| tableName, |
| isDecimalType ? "does not contain" : "contains unexpected", |
| nameType.getKey() |
| ) |
| ); |
| } |
| } |
| } |
