You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2021/11/01 23:16:19 UTC
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
From Wael Alkowaileet <wa...@gmail.com>:
Wael Alkowaileet has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
[ASTERIXDB-2981][EXT] Handle Parquet specialized types
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Handle parquet specialized types:
- Supported types: Date, Timestamp, Binary
- Unsupported types: Decimal
Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
---
A asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
M asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
31 files changed, 567 insertions(+), 131 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/03/13903/1
diff --git a/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet b/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
new file mode 100644
index 0000000..5a763fe
--- /dev/null
+++ b/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
Binary files differ
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
index d3865d3..bf173e5 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
@@ -60,6 +60,9 @@
//Write parquet files
File[] listOfFiles = srcPath.listFiles();
for (File jsonFile : listOfFiles) {
+ if (jsonFile.isDirectory()) {
+ continue;
+ }
String fileName = jsonFile.getName().substring(0, jsonFile.getName().indexOf(".")) + ".parquet";
Path outputPath = new Path(destPath.getAbsolutePath(), fileName);
writeParquetFile(jsonFile, outputPath);
@@ -67,10 +70,10 @@
}
private static void writeParquetFile(File jsonInputPath, Path parquetOutputPath) throws IOException {
- final FileInputStream schemaInputStream = new FileInputStream(jsonInputPath);
- final FileInputStream jsonInputStream = new FileInputStream(jsonInputPath);
+ FileInputStream schemaInputStream = new FileInputStream(jsonInputPath);
+ FileInputStream jsonInputStream = new FileInputStream(jsonInputPath);
//Infer Avro schema
- final Schema inputSchema = JsonUtil.inferSchema(schemaInputStream, "parquet_schema", NUM_OF_RECORDS_SCHEMA);
+ Schema inputSchema = JsonUtil.inferSchema(schemaInputStream, "parquet_schema", NUM_OF_RECORDS_SCHEMA);
try (JSONFileReader<Record> reader = new JSONFileReader<>(jsonInputStream, inputSchema, Record.class)) {
reader.initialize();
try (AvroParquetWriter<Record> writer = new AvroParquetWriter<>(parquetOutputPath, inputSchema)) {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
index f6b501a..c634d0b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
@@ -19,6 +19,7 @@
package org.apache.asterix.test.external_dataset;
import static org.apache.asterix.test.external_dataset.BinaryFileConverterUtil.BINARY_GEN_BASEDIR;
+import static org.apache.asterix.test.external_dataset.BinaryFileConverterUtil.DEFAULT_PARQUET_SRC_PATH;
import static org.apache.asterix.test.external_dataset.aws.AwsS3ExternalDatasetTest.FIXED_DATA_CONTAINER;
import java.io.BufferedWriter;
@@ -230,18 +231,23 @@
}
private static void loadParquetFiles() {
- String dataBasePath = BINARY_GEN_BASEDIR;
+ String generatedDataBasePath = BINARY_GEN_BASEDIR;
+ String providedDataBasePath = DEFAULT_PARQUET_SRC_PATH;
+ String binarySegment = "binary";
+
String definition = PARQUET_DEFINITION;
// Normal format
String definitionSegment = "";
- loadData(dataBasePath, "", "dummy_tweet.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_age.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_age-string.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_name.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_name_comment.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "heterogeneous_1.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "heterogeneous_2.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "dummy_tweet.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_age.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_age-string.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_name.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_name_comment.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "heterogeneous_1.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "heterogeneous_2.parquet", definition, definitionSegment, false, false);
+ loadData(providedDataBasePath, binarySegment, "parquetTypes.parquet", definition, definitionSegment, false,
+ false);
}
private static void loadData(String fileBasePath, String filePathSegment, String filename, String definition,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
new file mode 100644
index 0000000..264159e
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+/*
+* Description : Test Parquet types
+* Expected Res : Success with Warning (Decimal type is not supported)
+* Date : November 1st 2021
+*/
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+
+CREATE TYPE ParquetType as {
+};
+
+CREATE EXTERNAL DATASET ParquetDataset(ParquetType) USING %adapter%
+(
+ %template%,
+ ("container"="playground"),
+ ("definition"="parquet-data/reviews"),
+ ("include"="*parquetTypes.parquet"),
+ ("format" = "parquet")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
new file mode 100644
index 0000000..888d503
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+/*
+* Description : Test Parquet types
+* Expected Res : Success with Warning (Decimal type is not supported)
+* Date : November 1st 2021
+*/
+
+
+-- param max-warnings:json=1000
+
+USE test;
+
+SELECT VALUE p
+FROM ParquetDataset p
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
new file mode 100644
index 0000000..eb87c73
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
@@ -0,0 +1 @@
+{ "byteType": 1, "shortType": 1, "intType": 1, "longType": 1, "floatType": 1.0, "doubleType": 1.0, "binaryType": hex("000101"), "timestampType": datetime("2021-11-01T21:37:13.738"), "dateType": date("2021-11-01") }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
index 96d34ec..a97a6ca 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
@@ -134,6 +134,13 @@
<expected-warn>ASX0002: Type mismatch: function field-access-by-name expects its 1st input parameter to be of type object, but the actual input type is array (in line 31, at column 52)</expected-warn>
</compilation-unit>
</test-case>
+ <test-case FilePath="external-dataset" check-warnings="true">
+ <compilation-unit name="common/parquet/parquet-types">
+ <placeholder name="adapter" value="S3" />
+ <output-dir compare="Text">common/parquet/parquet-types</output-dir>
+ <expected-warn>ASX0053: Unsupported Parquet type "optional fixed_len_byte_array(16) decimalType (DECIMAL(38,18))</expected-warn>
+ </compilation-unit>
+ </test-case>
<test-case FilePath="external-dataset">
<compilation-unit name="common/parquet/heterogeneous-access-pushdown">
<placeholder name="adapter" value="S3" />
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index d49dffc..e50fdf9 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -80,6 +80,7 @@
INVALID_PARAM(50),
INCOMPARABLE_TYPES(51),
ILLEGAL_STATE(52),
+ UNSUPPORTED_PARQUET_TYPE(53),
UNSUPPORTED_JRE(100),
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index 136e169..a39be03 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -87,6 +87,7 @@
50 = Invalid parameter \"%1$s\"
#51 is used
52 = Illegal state. %1$s
+53 = Unsupported Parquet type "%1$s"
100 = Unsupported JRE: %1$s
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
index f15f735..6798cec 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
@@ -86,16 +86,16 @@
@Override
public void configure(IServiceContext serviceCtx, Map<String, String> configuration,
IWarningCollector warningCollector) throws AlgebricksException, HyracksDataException {
- JobConf hdfsConf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf hdfsConf = createHdfsConf(serviceCtx, configuration);
configureHdfsConf(hdfsConf, configuration);
}
- protected JobConf createHdfsConf(IServiceContext serviceCtx, Map<String, String> configuration, boolean shouldWarn)
+ protected JobConf createHdfsConf(IServiceContext serviceCtx, Map<String, String> configuration)
throws HyracksDataException {
this.serviceCtx = serviceCtx;
this.configuration = configuration;
init((ICCServiceContext) serviceCtx);
- return HDFSUtils.configureHDFSJobConf(configuration, shouldWarn);
+ return HDFSUtils.configureHDFSJobConf(configuration);
}
protected void configureHdfsConf(JobConf conf, Map<String, String> configuration) throws AlgebricksException {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
index 85d8671..803e657 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
@@ -50,7 +50,7 @@
putS3ConfToHadoopConf(configuration, path);
//Configure Hadoop S3 input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf conf = createHdfsConf(serviceCtx, configuration);
int numberOfPartitions = getPartitionConstraint().getLocations().length;
ExternalDataUtils.AwsS3.configureAwsS3HdfsJobConf(conf, configuration, numberOfPartitions);
configureHdfsConf(conf, configuration);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
index 0f9f484..ee765ce 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
@@ -54,7 +54,7 @@
putAzureBlobConfToHadoopConf(configuration, path);
//Configure Hadoop Azure input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf conf = createHdfsConf(serviceCtx, configuration);
ExternalDataUtils.Azure.configureAzureHdfsJobConf(conf, configuration, endPoint);
configureHdfsConf(conf, configuration);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
index 8fbe5c4..75d431d 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
@@ -65,7 +65,7 @@
IWarningCollector warningCollector) throws AsterixException {
this.serviceCtx = serviceCtx;
this.configuration = configuration;
- JobConf conf = HDFSUtils.configureHDFSJobConf(configuration, warningCollector.shouldWarn());
+ JobConf conf = HDFSUtils.configureHDFSJobConf(configuration);
try {
confFactory = new ConfFactory(conf);
} catch (HyracksDataException e) {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
index 363d2d2..26daeb9 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
@@ -19,12 +19,15 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IMutableValueStorage;
import org.apache.hyracks.data.std.api.IValueReference;
import org.apache.parquet.io.api.Converter;
import org.apache.parquet.io.api.GroupConverter;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.Type;
@@ -38,12 +41,13 @@
protected final ParserContext context;
protected IMutableValueStorage tempStorage;
- AbstractComplexConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- this(parent, null, index, parquetType, context);
+ AbstractComplexConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ this(parent, null, index, parquetType, context, warnings);
}
AbstractComplexConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
- GroupType parquetType, ParserContext context) {
+ GroupType parquetType, ParserContext context, List<Warning> warnings) {
this.parent = parent;
this.fieldName = fieldName;
this.index = index;
@@ -51,16 +55,14 @@
converters = new Converter[parquetType.getFieldCount()];
for (int i = 0; i < parquetType.getFieldCount(); i++) {
final Type type = parquetType.getType(i);
- if (type == AsterixTypeToParquetTypeVisitor.MISSING) {
- converters[i] = MissingConverter.INSTANCE;
- } else if (type.isPrimitive()) {
- converters[i] = createAtomicConverter(parquetType, i);
+ if (type.isPrimitive()) {
+ converters[i] = createAtomicConverter(parquetType, i, warnings);
} else if (LogicalTypeAnnotation.listType().equals(type.getLogicalTypeAnnotation())) {
- converters[i] = createArrayConverter(parquetType, i);
+ converters[i] = createArrayConverter(parquetType, i, warnings);
} else if (type.getRepetition() == Repetition.REPEATED) {
- converters[i] = createRepeatedConverter(parquetType, i);
+ converters[i] = createRepeatedConverter(parquetType, i, warnings);
} else {
- converters[i] = createObjectConverter(parquetType, i);
+ converters[i] = createObjectConverter(parquetType, i, warnings);
}
}
}
@@ -70,13 +72,13 @@
*
* @param value Child value
*/
- protected abstract void addValue(IFieldValue value);
+ public abstract void addValue(IFieldValue value);
- protected abstract AtomicConverter createAtomicConverter(GroupType type, int index);
+ protected abstract PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings);
- protected abstract ArrayConverter createArrayConverter(GroupType type, int index);
+ protected abstract ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings);
- protected abstract ObjectConverter createObjectConverter(GroupType type, int index);
+ protected abstract ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings);
/**
* Parquet file created by (old) Avro writer treat repeated values differently from files created by Spark.
@@ -106,13 +108,13 @@
*
* @formatter:on
*/
- private AbstractComplexConverter createRepeatedConverter(GroupType type, int index) {
+ private AbstractComplexConverter createRepeatedConverter(GroupType type, int index, List<Warning> warnings) {
GroupType repeatedType = type.getType(index).asGroupType();
//The name "array" is used by Avro to represent group element (array of objects)
if (repeatedType.getFieldCount() > 1 || "array".equals(repeatedType.getName())) {
- return new ObjectConverter(this, index, repeatedType, context);
+ return new ObjectConverter(this, index, repeatedType, context, warnings);
}
- return new RepeatedConverter(this, index, repeatedType, context);
+ return new RepeatedConverter(this, index, repeatedType, context, warnings);
}
@Override
@@ -130,7 +132,7 @@
return converters[fieldIndex];
}
- protected DataOutput getDataOutput() {
+ public DataOutput getDataOutput() {
tempStorage.reset();
return tempStorage.getDataOutput();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
index 9e8da77..81d7f0c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
@@ -19,24 +19,30 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
+import java.util.List;
import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class ArrayConverter extends AbstractComplexConverter {
private IAsterixListBuilder builder;
- public ArrayConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public ArrayConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
public ArrayConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, GroupType parquetType,
- ParserContext context) {
- super(parent, fieldName, index, parquetType, context);
+ ParserContext context, List<Warning> warnings) {
+ super(parent, fieldName, index, parquetType, context, warnings);
}
@Override
@@ -59,7 +65,7 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
try {
builder.addItem(tempStorage);
} catch (HyracksDataException e) {
@@ -68,19 +74,20 @@
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
- return new AtomicConverter(this, index, context);
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, index, context, warnings);
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, index, arrayType, context);
+ return new ArrayConverter(this, index, arrayType, context, warnings);
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType objectType = type.getType(index).asGroupType();
- return new ObjectConverter(this, index, objectType, context);
+ return new ObjectConverter(this, index, objectType, context, warnings);
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
index d596493..5212422 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
@@ -18,10 +18,13 @@
*/
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+import static org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider.MISSING;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.AUnionType;
@@ -33,6 +36,9 @@
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
@@ -43,14 +49,12 @@
*/
public class AsterixTypeToParquetTypeVisitor implements IATypeVisitor<Type, Type> {
public static final MessageType EMPTY_PARQUET_MESSAGE = Types.buildMessage().named("EMPTY");
- public static final PrimitiveType MISSING =
- Types.optional(PrimitiveType.PrimitiveTypeName.BOOLEAN).named("MISSING");
private final List<Warning> warnings;
private Map<String, FunctionCallInformation> funcInfo;
- public AsterixTypeToParquetTypeVisitor(boolean shouldWarn) {
- warnings = shouldWarn ? new ArrayList<>() : null;
+ public AsterixTypeToParquetTypeVisitor() {
+ warnings = new ArrayList<>();
}
public MessageType clipType(ARecordType rootType, MessageType fileSchema,
@@ -161,11 +165,17 @@
ATypeTag expectedType = node.getTypeTag();
boolean isNotExpected = actualType != expectedType;
- if (warnings != null && isNotExpected) {
+ if (isNotExpected) {
//typeName is unique
FunctionCallInformation info = funcInfo.get(node.getTypeName());
//If no warning is created, then it means it has been reported
- Warning warning = info.createTypeMismatchWarning(expectedType, actualType);
+ Warning warning;
+ if (actualType == ATypeTag.ANY) {
+ //Unsupported parquet type
+ warning = info.createWarning(ErrorCode.UNSUPPORTED_PARQUET_TYPE, type.toString());
+ } else {
+ warning = info.createTypeMismatchWarning(expectedType, actualType);
+ }
if (warning != null) {
//New warning that we saw for the first time. We should report it.
warnings.add(warning);
@@ -174,28 +184,31 @@
return isNotExpected;
}
- private static ATypeTag mapType(Type parquetType) {
+ public static ATypeTag mapType(Type parquetType) {
LogicalTypeAnnotation typeAnnotation = parquetType.getLogicalTypeAnnotation();
if (!parquetType.isPrimitive() && typeAnnotation == null) {
return ATypeTag.OBJECT;
} else if (typeAnnotation == LogicalTypeAnnotation.listType()) {
return ATypeTag.ARRAY;
- } else if (typeAnnotation == LogicalTypeAnnotation.stringType()) {
- return ATypeTag.STRING;
} else {
//Check other primitive types
- PrimitiveType.PrimitiveTypeName primitiveTypeName = parquetType.asPrimitiveType().getPrimitiveTypeName();
- switch (primitiveTypeName) {
+ PrimitiveType primitiveType = parquetType.asPrimitiveType();
+ switch (primitiveType.getPrimitiveTypeName()) {
case BOOLEAN:
return ATypeTag.BOOLEAN;
- case INT32:
- case INT64:
- return ATypeTag.BIGINT;
case FLOAT:
case DOUBLE:
return ATypeTag.DOUBLE;
+ case INT32:
+ case INT64:
+ return handleInt(primitiveType);
+ case INT96:
+ return ATypeTag.DATETIME;
+ case BINARY:
+ return handleBinary(primitiveType);
default:
- throw new IllegalStateException("Unsupported type " + parquetType);
+ //Use ANY for unsupported types
+ return ATypeTag.ANY;
}
}
}
@@ -206,4 +219,35 @@
}
return MISSING;
}
+
+ private static ATypeTag handleInt(PrimitiveType type) {
+ LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+ if (logicalType == null) {
+ return ATypeTag.BIGINT;
+ } else if (logicalType instanceof IntLogicalTypeAnnotation) {
+ IntLogicalTypeAnnotation intLogicalType = (IntLogicalTypeAnnotation) logicalType;
+ if (intLogicalType.getBitWidth() == 64 && !intLogicalType.isSigned()) {
+ //Overflow
+ return ATypeTag.ANY;
+ }
+ return ATypeTag.BIGINT;
+ } else if (logicalType instanceof DateLogicalTypeAnnotation) {
+ return ATypeTag.DATE;
+ } else if (logicalType instanceof TimestampLogicalTypeAnnotation) {
+ return ATypeTag.DATETIME;
+ }
+ //Decimal type
+ return ATypeTag.ANY;
+ }
+
+ private static ATypeTag handleBinary(PrimitiveType type) {
+ LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+ if (logicalType == null) {
+ return ATypeTag.BINARY;
+ } else if (logicalType == LogicalTypeAnnotation.stringType() || logicalType == LogicalTypeAnnotation.enumType()
+ || logicalType == LogicalTypeAnnotation.jsonType()) {
+ return ATypeTag.STRING;
+ }
+ return ATypeTag.ANY;
+ }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
index 946b10c..1c8a167 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
@@ -24,7 +24,7 @@
* This interface is intended to extend {@link org.apache.parquet.io.api.Converter} to get the field name or the
* index of associated with a value.
*/
-interface IFieldValue {
+public interface IFieldValue {
IValueReference getFieldName();
/**
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
index 8736e7d..c1391dc 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
@@ -19,24 +19,30 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
+import java.util.List;
import org.apache.asterix.builders.IARecordBuilder;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class ObjectConverter extends AbstractComplexConverter {
private IARecordBuilder builder;
- public ObjectConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public ObjectConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
public ObjectConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, GroupType parquetType,
- ParserContext context) {
- super(parent, fieldName, index, parquetType, context);
+ ParserContext context, List<Warning> warnings) {
+ super(parent, fieldName, index, parquetType, context, warnings);
}
@Override
@@ -59,7 +65,7 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
try {
builder.addField(value.getFieldName(), getValue());
} catch (HyracksDataException e) {
@@ -68,31 +74,34 @@
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
try {
- return new AtomicConverter(this, context.getSerializedFieldName(type.getFieldName(index)), index, context);
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ IValueReference fieldName = context.getSerializedFieldName(type.getFieldName(index));
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, fieldName, index, context,
+ warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
try {
final IValueReference childFieldName = context.getSerializedFieldName(type.getFieldName(index));
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, childFieldName, index, arrayType, context);
+ return new ArrayConverter(this, childFieldName, index, arrayType, context, warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
try {
final IValueReference childFieldName = context.getSerializedFieldName(type.getFieldName(index));
final GroupType objectType = type.getType(index).asGroupType();
- return new ObjectConverter(this, childFieldName, index, objectType, context);
+ return new ObjectConverter(this, childFieldName, index, objectType, context, warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
index b06afd3..3e4d811 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
@@ -51,10 +51,8 @@
@Override
public void close() throws IOException {
super.close();
- if (warningCollector.shouldWarn()) {
- //report warnings
- HDFSUtils.issueWarnings(warningCollector, conf);
- }
+ //report warnings
+ HDFSUtils.issueWarnings(warningCollector, conf);
}
@SuppressWarnings("unchecked")
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
index b2a5eeb..2f219e0 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
@@ -19,11 +19,10 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.HDFSUtils;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.runtime.projection.FunctionCallInformation;
@@ -40,20 +39,19 @@
@Override
public ReadContext init(InitContext context) {
MessageType requestedSchema = getRequestedSchema(context);
- return new ReadContext(requestedSchema, Collections.emptyMap());
+ return new ReadContext(requestedSchema);
}
@Override
public RecordMaterializer<IValueReference> prepareForRead(Configuration configuration,
Map<String, String> keyValueMetaData, MessageType fileSchema, ReadContext readContext) {
- return new ADMRecordMaterializer(readContext);
+ return new ADMRecordMaterializer(configuration, readContext);
}
private static MessageType getRequestedSchema(InitContext context) {
Configuration configuration = context.getConfiguration();
MessageType fileSchema = context.getFileSchema();
- boolean shouldWarn = configuration.getBoolean(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_ENABLED, false);
- AsterixTypeToParquetTypeVisitor visitor = new AsterixTypeToParquetTypeVisitor(shouldWarn);
+ AsterixTypeToParquetTypeVisitor visitor = new AsterixTypeToParquetTypeVisitor();
try {
ARecordType expectedType = HDFSUtils.getExpectedType(configuration);
Map<String, FunctionCallInformation> functionCallInformationMap =
@@ -61,7 +59,7 @@
MessageType requestedType = visitor.clipType(expectedType, fileSchema, functionCallInformationMap);
List<Warning> warnings = visitor.getWarnings();
- if (shouldWarn && !warnings.isEmpty()) {
+ if (!warnings.isEmpty()) {
//New warnings were created, set the warnings in hadoop configuration to be reported
HDFSUtils.setWarnings(warnings, configuration);
//Update the reported warnings so that we do not report the same warning again
@@ -76,8 +74,17 @@
private static class ADMRecordMaterializer extends RecordMaterializer<IValueReference> {
private final RootConverter rootConverter;
- public ADMRecordMaterializer(ReadContext readContext) {
- rootConverter = new RootConverter(readContext.getRequestedSchema());
+ public ADMRecordMaterializer(Configuration configuration, ReadContext readContext) {
+ List<Warning> warnings = new ArrayList<>();
+ rootConverter = new RootConverter(readContext.getRequestedSchema(), warnings);
+ try {
+ if (!warnings.isEmpty()) {
+ //Issue warnings if encountered an unsupported type
+ HDFSUtils.setWarnings(warnings, configuration);
+ }
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
index d46d84c..b249f8b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
@@ -19,13 +19,19 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class RepeatedConverter extends AbstractComplexConverter {
- public RepeatedConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public RepeatedConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
@Override
@@ -39,28 +45,29 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
parent.addValue(value);
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
- return new AtomicConverter(this, index, context);
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, index, context, warnings);
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, index, arrayType, context);
+ return new ArrayConverter(this, index, arrayType, context, warnings);
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
- return new ObjectConverter(this, index, type.getType(index).asGroupType(), context);
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
+ return new ObjectConverter(this, index, type.getType(index).asGroupType(), context, warnings);
}
@Override
- protected DataOutput getDataOutput() {
+ public DataOutput getDataOutput() {
return getParentDataOutput();
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
index 76f4342..d7f8018 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
@@ -19,8 +19,10 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.parquet.schema.GroupType;
@@ -28,8 +30,8 @@
class RootConverter extends ObjectConverter {
private final ArrayBackedValueStorage rootBuffer;
- public RootConverter(GroupType parquetType) {
- super(null, -1, parquetType, new ParserContext(true));
+ public RootConverter(GroupType parquetType, List<Warning> warnings) {
+ super(null, -1, parquetType, new ParserContext(true), warnings);
this.rootBuffer = new ArrayBackedValueStorage();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
new file mode 100644
index 0000000..9506e07
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
@@ -0,0 +1,59 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder;
+import org.apache.parquet.io.api.Binary;
+
+public class BinaryConverter extends GenericPrimitiveConverter {
+ private byte[] lengthBytes;
+
+ BinaryConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ @Override
+ public void addBinary(Binary value) {
+ DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.BINARY.serialize());
+ writeLength(value.length(), out);
+ value.writeTo(out);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ private void writeLength(int length, DataOutput out) throws IOException {
+ int requiredLength = VarLenIntEncoderDecoder.getBytesRequired(length);
+ if (lengthBytes == null || requiredLength > lengthBytes.length) {
+ lengthBytes = new byte[requiredLength];
+ }
+ VarLenIntEncoderDecoder.encode(length, lengthBytes, 0);
+ out.write(lengthBytes, 0, requiredLength);
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
new file mode 100644
index 0000000..f33865c
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+
+class DateConverter extends GenericPrimitiveConverter {
+ DateConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ @Override
+ public void addInt(int value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATE.serialize());
+ out.writeInt(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
similarity index 83%
rename from asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java
rename to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
index 7219bdd..88b65ef 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
@@ -16,11 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+package org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve;
import java.io.DataOutput;
import java.io.IOException;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.IFieldValue;
import org.apache.asterix.external.input.stream.StandardUTF8ToModifiedUTF8DataOutput;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.types.ATypeTag;
@@ -28,20 +30,14 @@
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.io.api.PrimitiveConverter;
-/**
- * Currently, only JSON types are supported (string, number, boolean)
- */
-class AtomicConverter extends PrimitiveConverter implements IFieldValue {
- private final AbstractComplexConverter parent;
- private final IValueReference fieldName;
- private final int index;
- private final ParserContext context;
+public class GenericPrimitiveConverter extends PrimitiveConverter implements IFieldValue {
- public AtomicConverter(AbstractComplexConverter parent, int index, ParserContext context) {
- this(parent, null, index, context);
- }
+ protected final AbstractComplexConverter parent;
+ protected final IValueReference fieldName;
+ protected final int index;
+ protected final ParserContext context;
- public AtomicConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
+ GenericPrimitiveConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
ParserContext context) {
this.parent = parent;
this.fieldName = fieldName;
@@ -50,6 +46,16 @@
}
@Override
+ public final IValueReference getFieldName() {
+ return fieldName;
+ }
+
+ @Override
+ public final int getIndex() {
+ return index;
+ }
+
+ @Override
public void addBinary(Binary value) {
final DataOutput out = parent.getDataOutput();
final StandardUTF8ToModifiedUTF8DataOutput stringOut = context.getModifiedUTF8DataOutput();
@@ -108,14 +114,4 @@
}
parent.addValue(this);
}
-
- @Override
- public IValueReference getFieldName() {
- return fieldName;
- }
-
- @Override
- public int getIndex() {
- return index;
- }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
similarity index 98%
rename from asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java
rename to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
index e38056b..21b9217 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+package org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.io.api.PrimitiveConverter;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
new file mode 100644
index 0000000..9fe2b42
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
@@ -0,0 +1,75 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.util.List;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AsterixTypeToParquetTypeVisitor;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.api.exceptions.Warning;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Types;
+
+public class PrimitiveConverterProvider {
+ public static final PrimitiveType MISSING =
+ Types.optional(PrimitiveType.PrimitiveTypeName.BOOLEAN).named("MISSING");
+
+ private PrimitiveConverterProvider() {
+ }
+
+ public static PrimitiveConverter createPrimitiveConverter(PrimitiveType type, AbstractComplexConverter parent,
+ int index, ParserContext context, List<Warning> warnings) {
+ return createPrimitiveConverter(type, parent, null, index, context, warnings);
+ }
+
+ public static PrimitiveConverter createPrimitiveConverter(PrimitiveType type, AbstractComplexConverter parent,
+ IValueReference fieldName, int index, ParserContext context, List<Warning> warnings) {
+
+ if (type == MISSING) {
+ return MissingConverter.INSTANCE;
+ }
+
+ ATypeTag mappedType = AsterixTypeToParquetTypeVisitor.mapType(type);
+ switch (mappedType) {
+ case BOOLEAN:
+ case BIGINT:
+ case DOUBLE:
+ case STRING:
+ return new GenericPrimitiveConverter(parent, fieldName, index, context);
+ case BINARY:
+ return new BinaryConverter(parent, fieldName, index, context);
+ case DATE:
+ return new DateConverter(parent, fieldName, index, context);
+ case DATETIME:
+ return new TimestampConverter(parent, fieldName, index, context);
+ default:
+ return warn(warnings, type);
+ }
+ }
+
+ private static PrimitiveConverter warn(List<Warning> warnings, PrimitiveType type) {
+ warnings.add(Warning.of(null, ErrorCode.UNSUPPORTED_PARQUET_TYPE, type.toString()));
+ return MissingConverter.INSTANCE;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
new file mode 100644
index 0000000..6e6fa09
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
@@ -0,0 +1,96 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.Binary;
+
+class TimestampConverter extends GenericPrimitiveConverter {
+ private static final long JULIAN_DAY_OF_EPOCH = 2440588;
+ private static final long MILLIS_PER_DAY = 86400000L;
+ private static final long NANOS_PER_MILLIS = 1000000L;
+
+ TimestampConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ /**
+ * Timestamp is an INT96 (Little Endian)
+ * Skip nanoseconds part
+ *
+ * @param value binary representation of INT96
+ */
+ @Override
+ public void addBinary(Binary value) {
+ DataOutput out = parent.getDataOutput();
+ ByteBuffer buffer = value.toByteBuffer().order(ByteOrder.LITTLE_ENDIAN);
+ long timeOfDayNanos = buffer.getLong();
+ int julianDay = buffer.getInt();
+ long timestamp = fromJulian(julianDay, timeOfDayNanos);
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(timestamp);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+
+ parent.addValue(this);
+ }
+
+ @Override
+ public void addInt(int value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ /**
+ * Timestamp is an INT64
+ *
+ * @param value long value
+ */
+ @Override
+ public void addLong(long value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ private static long fromJulian(int days, long nanos) {
+ return (days - JULIAN_DAY_OF_EPOCH) * MILLIS_PER_DAY + nanos / NANOS_PER_MILLIS;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index ec25997..16a0f66 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -60,7 +60,6 @@
public static final String KEY_HADOOP_SHORT_CIRCUIT = "dfs.client.read.shortcircuit";
public static final String KEY_HADOOP_SOCKET_PATH = "dfs.domain.socket.path";
public static final String KEY_HADOOP_BUFFER_SIZE = "io.file.buffer.size";
- public static final String KEY_HADOOP_ASTERIX_WARNINGS_ENABLED = "org.apache.asterix.warnings.enabled";
//Base64 encoded warnings issued from Hadoop
public static final String KEY_HADOOP_ASTERIX_WARNINGS_LIST = "org.apache.asterix.warnings.list";
//Disable caching FileSystem for Hadoop
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index 9f65cd7..25cd832 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -201,7 +201,7 @@
}
}
- public static JobConf configureHDFSJobConf(Map<String, String> configuration, boolean shouldWarn) {
+ public static JobConf configureHDFSJobConf(Map<String, String> configuration) {
JobConf conf = new JobConf();
String localShortCircuitSocketPath = configuration.get(ExternalDataConstants.KEY_LOCAL_SOCKET_PATH);
String formatClassName = HDFSUtils.getInputFormatClassName(configuration);
@@ -235,11 +235,6 @@
//Subset of the values were requested, set the functionCallInformation
conf.set(ExternalDataConstants.KEY_HADOOP_ASTERIX_FUNCTION_CALL_INFORMATION,
configuration.get(ExternalDataConstants.KEY_HADOOP_ASTERIX_FUNCTION_CALL_INFORMATION));
- /*
- * Allows Parquet to issue warnings in case we found type mismatches (if warnings are enabled).
- * Warnings will be issued during the type matching of Parquet's schema with the requested schema
- */
- conf.setBoolean(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_ENABLED, shouldWarn);
}
conf.set(ExternalDataConstants.KEY_REQUESTED_FIELDS, requestedValues);
}
@@ -320,10 +315,10 @@
public static void issueWarnings(IWarningCollector warningCollector, Configuration conf) throws IOException {
String warnings = conf.get(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_LIST, "");
- if (!warnings.isEmpty() && warningCollector.shouldWarn()) {
+ if (!warnings.isEmpty()) {
String[] encodedWarnings = warnings.split(",");
Base64.Decoder decoder = Base64.getDecoder();
- for (int i = 0; i < encodedWarnings.length; i++) {
+ for (int i = 0; i < encodedWarnings.length && warningCollector.shouldWarn(); i++) {
/*
* This should create a small number of objects as warnings are reported only once by AsterixDB's
* hadoop readers
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
index 5cb26fd..ee2f168 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
@@ -73,6 +73,10 @@
ExceptionUtil.indexToPosition(0), expectedType, actualType);
}
+ public Warning createWarning(ErrorCode errorCode, Serializable... args) {
+ return Warning.of(sourceLocation, errorCode, args);
+ }
+
public void writeFields(DataOutput output) throws IOException {
output.writeUTF(functionName);
SourceLocation.writeFields(sourceLocation, output);
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 1
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newchange
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Jenkins <je...@fulliautomatix.ics.uci.edu>:
Jenkins has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 5: Integration-Tests+1
Integration Tests Successful
https://asterix-jenkins.ics.uci.edu/job/asterix-gerrit-integration-tests/12862/ : SUCCESS
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 5
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wael Alkowaileet <wa...@gmail.com>
Gerrit-Comment-Date: Sun, 09 Jan 2022 22:45:02 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Wael Alkowaileet <wa...@gmail.com>:
Wael Alkowaileet has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
[ASTERIXDB-2981][EXT] Handle Parquet specialized types
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Handle parquet specialized types:
- Supported types: Date, Timestamp, Binary
- Unsupported types: Decimal
Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
---
A asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
M asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
31 files changed, 567 insertions(+), 131 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/03/13903/1
diff --git a/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet b/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
new file mode 100644
index 0000000..5a763fe
--- /dev/null
+++ b/asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
Binary files differ
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
index d3865d3..bf173e5 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
@@ -60,6 +60,9 @@
//Write parquet files
File[] listOfFiles = srcPath.listFiles();
for (File jsonFile : listOfFiles) {
+ if (jsonFile.isDirectory()) {
+ continue;
+ }
String fileName = jsonFile.getName().substring(0, jsonFile.getName().indexOf(".")) + ".parquet";
Path outputPath = new Path(destPath.getAbsolutePath(), fileName);
writeParquetFile(jsonFile, outputPath);
@@ -67,10 +70,10 @@
}
private static void writeParquetFile(File jsonInputPath, Path parquetOutputPath) throws IOException {
- final FileInputStream schemaInputStream = new FileInputStream(jsonInputPath);
- final FileInputStream jsonInputStream = new FileInputStream(jsonInputPath);
+ FileInputStream schemaInputStream = new FileInputStream(jsonInputPath);
+ FileInputStream jsonInputStream = new FileInputStream(jsonInputPath);
//Infer Avro schema
- final Schema inputSchema = JsonUtil.inferSchema(schemaInputStream, "parquet_schema", NUM_OF_RECORDS_SCHEMA);
+ Schema inputSchema = JsonUtil.inferSchema(schemaInputStream, "parquet_schema", NUM_OF_RECORDS_SCHEMA);
try (JSONFileReader<Record> reader = new JSONFileReader<>(jsonInputStream, inputSchema, Record.class)) {
reader.initialize();
try (AvroParquetWriter<Record> writer = new AvroParquetWriter<>(parquetOutputPath, inputSchema)) {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
index f6b501a..c634d0b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
@@ -19,6 +19,7 @@
package org.apache.asterix.test.external_dataset;
import static org.apache.asterix.test.external_dataset.BinaryFileConverterUtil.BINARY_GEN_BASEDIR;
+import static org.apache.asterix.test.external_dataset.BinaryFileConverterUtil.DEFAULT_PARQUET_SRC_PATH;
import static org.apache.asterix.test.external_dataset.aws.AwsS3ExternalDatasetTest.FIXED_DATA_CONTAINER;
import java.io.BufferedWriter;
@@ -230,18 +231,23 @@
}
private static void loadParquetFiles() {
- String dataBasePath = BINARY_GEN_BASEDIR;
+ String generatedDataBasePath = BINARY_GEN_BASEDIR;
+ String providedDataBasePath = DEFAULT_PARQUET_SRC_PATH;
+ String binarySegment = "binary";
+
String definition = PARQUET_DEFINITION;
// Normal format
String definitionSegment = "";
- loadData(dataBasePath, "", "dummy_tweet.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_age.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_age-string.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_name.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "id_name_comment.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "heterogeneous_1.parquet", definition, definitionSegment, false, false);
- loadData(dataBasePath, "", "heterogeneous_2.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "dummy_tweet.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_age.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_age-string.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_name.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "id_name_comment.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "heterogeneous_1.parquet", definition, definitionSegment, false, false);
+ loadData(generatedDataBasePath, "", "heterogeneous_2.parquet", definition, definitionSegment, false, false);
+ loadData(providedDataBasePath, binarySegment, "parquetTypes.parquet", definition, definitionSegment, false,
+ false);
}
private static void loadData(String fileBasePath, String filePathSegment, String filename, String definition,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
new file mode 100644
index 0000000..264159e
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+/*
+* Description : Test Parquet types
+* Expected Res : Success with Warning (Decimal type is not supported)
+* Date : November 1st 2021
+*/
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+
+CREATE TYPE ParquetType as {
+};
+
+CREATE EXTERNAL DATASET ParquetDataset(ParquetType) USING %adapter%
+(
+ %template%,
+ ("container"="playground"),
+ ("definition"="parquet-data/reviews"),
+ ("include"="*parquetTypes.parquet"),
+ ("format" = "parquet")
+);
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
new file mode 100644
index 0000000..888d503
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+/*
+* Description : Test Parquet types
+* Expected Res : Success with Warning (Decimal type is not supported)
+* Date : November 1st 2021
+*/
+
+
+-- param max-warnings:json=1000
+
+USE test;
+
+SELECT VALUE p
+FROM ParquetDataset p
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
new file mode 100644
index 0000000..eb87c73
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
@@ -0,0 +1 @@
+{ "byteType": 1, "shortType": 1, "intType": 1, "longType": 1, "floatType": 1.0, "doubleType": 1.0, "binaryType": hex("000101"), "timestampType": datetime("2021-11-01T21:37:13.738"), "dateType": date("2021-11-01") }
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
index 96d34ec..a97a6ca 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
@@ -134,6 +134,13 @@
<expected-warn>ASX0002: Type mismatch: function field-access-by-name expects its 1st input parameter to be of type object, but the actual input type is array (in line 31, at column 52)</expected-warn>
</compilation-unit>
</test-case>
+ <test-case FilePath="external-dataset" check-warnings="true">
+ <compilation-unit name="common/parquet/parquet-types">
+ <placeholder name="adapter" value="S3" />
+ <output-dir compare="Text">common/parquet/parquet-types</output-dir>
+ <expected-warn>ASX0053: Unsupported Parquet type "optional fixed_len_byte_array(16) decimalType (DECIMAL(38,18))</expected-warn>
+ </compilation-unit>
+ </test-case>
<test-case FilePath="external-dataset">
<compilation-unit name="common/parquet/heterogeneous-access-pushdown">
<placeholder name="adapter" value="S3" />
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index d49dffc..e50fdf9 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -80,6 +80,7 @@
INVALID_PARAM(50),
INCOMPARABLE_TYPES(51),
ILLEGAL_STATE(52),
+ UNSUPPORTED_PARQUET_TYPE(53),
UNSUPPORTED_JRE(100),
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index 136e169..a39be03 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -87,6 +87,7 @@
50 = Invalid parameter \"%1$s\"
#51 is used
52 = Illegal state. %1$s
+53 = Unsupported Parquet type "%1$s"
100 = Unsupported JRE: %1$s
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
index f15f735..6798cec 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/HDFSDataSourceFactory.java
@@ -86,16 +86,16 @@
@Override
public void configure(IServiceContext serviceCtx, Map<String, String> configuration,
IWarningCollector warningCollector) throws AlgebricksException, HyracksDataException {
- JobConf hdfsConf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf hdfsConf = createHdfsConf(serviceCtx, configuration);
configureHdfsConf(hdfsConf, configuration);
}
- protected JobConf createHdfsConf(IServiceContext serviceCtx, Map<String, String> configuration, boolean shouldWarn)
+ protected JobConf createHdfsConf(IServiceContext serviceCtx, Map<String, String> configuration)
throws HyracksDataException {
this.serviceCtx = serviceCtx;
this.configuration = configuration;
init((ICCServiceContext) serviceCtx);
- return HDFSUtils.configureHDFSJobConf(configuration, shouldWarn);
+ return HDFSUtils.configureHDFSJobConf(configuration);
}
protected void configureHdfsConf(JobConf conf, Map<String, String> configuration) throws AlgebricksException {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
index 85d8671..803e657 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/parquet/AwsS3ParquetReaderFactory.java
@@ -50,7 +50,7 @@
putS3ConfToHadoopConf(configuration, path);
//Configure Hadoop S3 input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf conf = createHdfsConf(serviceCtx, configuration);
int numberOfPartitions = getPartitionConstraint().getLocations().length;
ExternalDataUtils.AwsS3.configureAwsS3HdfsJobConf(conf, configuration, numberOfPartitions);
configureHdfsConf(conf, configuration);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
index 0f9f484..ee765ce 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/parquet/AzureBlobParquetReaderFactory.java
@@ -54,7 +54,7 @@
putAzureBlobConfToHadoopConf(configuration, path);
//Configure Hadoop Azure input splits
- JobConf conf = createHdfsConf(serviceCtx, configuration, warningCollector.shouldWarn());
+ JobConf conf = createHdfsConf(serviceCtx, configuration);
ExternalDataUtils.Azure.configureAzureHdfsJobConf(conf, configuration, endPoint);
configureHdfsConf(conf, configuration);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
index 8fbe5c4..75d431d 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/HDFSLookupReaderFactory.java
@@ -65,7 +65,7 @@
IWarningCollector warningCollector) throws AsterixException {
this.serviceCtx = serviceCtx;
this.configuration = configuration;
- JobConf conf = HDFSUtils.configureHDFSJobConf(configuration, warningCollector.shouldWarn());
+ JobConf conf = HDFSUtils.configureHDFSJobConf(configuration);
try {
confFactory = new ConfFactory(conf);
} catch (HyracksDataException e) {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
index 363d2d2..26daeb9 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AbstractComplexConverter.java
@@ -19,12 +19,15 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IMutableValueStorage;
import org.apache.hyracks.data.std.api.IValueReference;
import org.apache.parquet.io.api.Converter;
import org.apache.parquet.io.api.GroupConverter;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.Type;
@@ -38,12 +41,13 @@
protected final ParserContext context;
protected IMutableValueStorage tempStorage;
- AbstractComplexConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- this(parent, null, index, parquetType, context);
+ AbstractComplexConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ this(parent, null, index, parquetType, context, warnings);
}
AbstractComplexConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
- GroupType parquetType, ParserContext context) {
+ GroupType parquetType, ParserContext context, List<Warning> warnings) {
this.parent = parent;
this.fieldName = fieldName;
this.index = index;
@@ -51,16 +55,14 @@
converters = new Converter[parquetType.getFieldCount()];
for (int i = 0; i < parquetType.getFieldCount(); i++) {
final Type type = parquetType.getType(i);
- if (type == AsterixTypeToParquetTypeVisitor.MISSING) {
- converters[i] = MissingConverter.INSTANCE;
- } else if (type.isPrimitive()) {
- converters[i] = createAtomicConverter(parquetType, i);
+ if (type.isPrimitive()) {
+ converters[i] = createAtomicConverter(parquetType, i, warnings);
} else if (LogicalTypeAnnotation.listType().equals(type.getLogicalTypeAnnotation())) {
- converters[i] = createArrayConverter(parquetType, i);
+ converters[i] = createArrayConverter(parquetType, i, warnings);
} else if (type.getRepetition() == Repetition.REPEATED) {
- converters[i] = createRepeatedConverter(parquetType, i);
+ converters[i] = createRepeatedConverter(parquetType, i, warnings);
} else {
- converters[i] = createObjectConverter(parquetType, i);
+ converters[i] = createObjectConverter(parquetType, i, warnings);
}
}
}
@@ -70,13 +72,13 @@
*
* @param value Child value
*/
- protected abstract void addValue(IFieldValue value);
+ public abstract void addValue(IFieldValue value);
- protected abstract AtomicConverter createAtomicConverter(GroupType type, int index);
+ protected abstract PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings);
- protected abstract ArrayConverter createArrayConverter(GroupType type, int index);
+ protected abstract ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings);
- protected abstract ObjectConverter createObjectConverter(GroupType type, int index);
+ protected abstract ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings);
/**
* Parquet file created by (old) Avro writer treat repeated values differently from files created by Spark.
@@ -106,13 +108,13 @@
*
* @formatter:on
*/
- private AbstractComplexConverter createRepeatedConverter(GroupType type, int index) {
+ private AbstractComplexConverter createRepeatedConverter(GroupType type, int index, List<Warning> warnings) {
GroupType repeatedType = type.getType(index).asGroupType();
//The name "array" is used by Avro to represent group element (array of objects)
if (repeatedType.getFieldCount() > 1 || "array".equals(repeatedType.getName())) {
- return new ObjectConverter(this, index, repeatedType, context);
+ return new ObjectConverter(this, index, repeatedType, context, warnings);
}
- return new RepeatedConverter(this, index, repeatedType, context);
+ return new RepeatedConverter(this, index, repeatedType, context, warnings);
}
@Override
@@ -130,7 +132,7 @@
return converters[fieldIndex];
}
- protected DataOutput getDataOutput() {
+ public DataOutput getDataOutput() {
tempStorage.reset();
return tempStorage.getDataOutput();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
index 9e8da77..81d7f0c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ArrayConverter.java
@@ -19,24 +19,30 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
+import java.util.List;
import org.apache.asterix.builders.IAsterixListBuilder;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class ArrayConverter extends AbstractComplexConverter {
private IAsterixListBuilder builder;
- public ArrayConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public ArrayConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
public ArrayConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, GroupType parquetType,
- ParserContext context) {
- super(parent, fieldName, index, parquetType, context);
+ ParserContext context, List<Warning> warnings) {
+ super(parent, fieldName, index, parquetType, context, warnings);
}
@Override
@@ -59,7 +65,7 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
try {
builder.addItem(tempStorage);
} catch (HyracksDataException e) {
@@ -68,19 +74,20 @@
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
- return new AtomicConverter(this, index, context);
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, index, context, warnings);
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, index, arrayType, context);
+ return new ArrayConverter(this, index, arrayType, context, warnings);
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType objectType = type.getType(index).asGroupType();
- return new ObjectConverter(this, index, objectType, context);
+ return new ObjectConverter(this, index, objectType, context, warnings);
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
index d596493..5212422 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
@@ -18,10 +18,13 @@
*/
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+import static org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider.MISSING;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.AUnionType;
@@ -33,6 +36,9 @@
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
@@ -43,14 +49,12 @@
*/
public class AsterixTypeToParquetTypeVisitor implements IATypeVisitor<Type, Type> {
public static final MessageType EMPTY_PARQUET_MESSAGE = Types.buildMessage().named("EMPTY");
- public static final PrimitiveType MISSING =
- Types.optional(PrimitiveType.PrimitiveTypeName.BOOLEAN).named("MISSING");
private final List<Warning> warnings;
private Map<String, FunctionCallInformation> funcInfo;
- public AsterixTypeToParquetTypeVisitor(boolean shouldWarn) {
- warnings = shouldWarn ? new ArrayList<>() : null;
+ public AsterixTypeToParquetTypeVisitor() {
+ warnings = new ArrayList<>();
}
public MessageType clipType(ARecordType rootType, MessageType fileSchema,
@@ -161,11 +165,17 @@
ATypeTag expectedType = node.getTypeTag();
boolean isNotExpected = actualType != expectedType;
- if (warnings != null && isNotExpected) {
+ if (isNotExpected) {
//typeName is unique
FunctionCallInformation info = funcInfo.get(node.getTypeName());
//If no warning is created, then it means it has been reported
- Warning warning = info.createTypeMismatchWarning(expectedType, actualType);
+ Warning warning;
+ if (actualType == ATypeTag.ANY) {
+ //Unsupported parquet type
+ warning = info.createWarning(ErrorCode.UNSUPPORTED_PARQUET_TYPE, type.toString());
+ } else {
+ warning = info.createTypeMismatchWarning(expectedType, actualType);
+ }
if (warning != null) {
//New warning that we saw for the first time. We should report it.
warnings.add(warning);
@@ -174,28 +184,31 @@
return isNotExpected;
}
- private static ATypeTag mapType(Type parquetType) {
+ public static ATypeTag mapType(Type parquetType) {
LogicalTypeAnnotation typeAnnotation = parquetType.getLogicalTypeAnnotation();
if (!parquetType.isPrimitive() && typeAnnotation == null) {
return ATypeTag.OBJECT;
} else if (typeAnnotation == LogicalTypeAnnotation.listType()) {
return ATypeTag.ARRAY;
- } else if (typeAnnotation == LogicalTypeAnnotation.stringType()) {
- return ATypeTag.STRING;
} else {
//Check other primitive types
- PrimitiveType.PrimitiveTypeName primitiveTypeName = parquetType.asPrimitiveType().getPrimitiveTypeName();
- switch (primitiveTypeName) {
+ PrimitiveType primitiveType = parquetType.asPrimitiveType();
+ switch (primitiveType.getPrimitiveTypeName()) {
case BOOLEAN:
return ATypeTag.BOOLEAN;
- case INT32:
- case INT64:
- return ATypeTag.BIGINT;
case FLOAT:
case DOUBLE:
return ATypeTag.DOUBLE;
+ case INT32:
+ case INT64:
+ return handleInt(primitiveType);
+ case INT96:
+ return ATypeTag.DATETIME;
+ case BINARY:
+ return handleBinary(primitiveType);
default:
- throw new IllegalStateException("Unsupported type " + parquetType);
+ //Use ANY for unsupported types
+ return ATypeTag.ANY;
}
}
}
@@ -206,4 +219,35 @@
}
return MISSING;
}
+
+ private static ATypeTag handleInt(PrimitiveType type) {
+ LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+ if (logicalType == null) {
+ return ATypeTag.BIGINT;
+ } else if (logicalType instanceof IntLogicalTypeAnnotation) {
+ IntLogicalTypeAnnotation intLogicalType = (IntLogicalTypeAnnotation) logicalType;
+ if (intLogicalType.getBitWidth() == 64 && !intLogicalType.isSigned()) {
+ //Overflow
+ return ATypeTag.ANY;
+ }
+ return ATypeTag.BIGINT;
+ } else if (logicalType instanceof DateLogicalTypeAnnotation) {
+ return ATypeTag.DATE;
+ } else if (logicalType instanceof TimestampLogicalTypeAnnotation) {
+ return ATypeTag.DATETIME;
+ }
+ //Decimal type
+ return ATypeTag.ANY;
+ }
+
+ private static ATypeTag handleBinary(PrimitiveType type) {
+ LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+ if (logicalType == null) {
+ return ATypeTag.BINARY;
+ } else if (logicalType == LogicalTypeAnnotation.stringType() || logicalType == LogicalTypeAnnotation.enumType()
+ || logicalType == LogicalTypeAnnotation.jsonType()) {
+ return ATypeTag.STRING;
+ }
+ return ATypeTag.ANY;
+ }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
index 946b10c..1c8a167 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/IFieldValue.java
@@ -24,7 +24,7 @@
* This interface is intended to extend {@link org.apache.parquet.io.api.Converter} to get the field name or the
* index of associated with a value.
*/
-interface IFieldValue {
+public interface IFieldValue {
IValueReference getFieldName();
/**
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
index 8736e7d..c1391dc 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ObjectConverter.java
@@ -19,24 +19,30 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
+import java.util.List;
import org.apache.asterix.builders.IARecordBuilder;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class ObjectConverter extends AbstractComplexConverter {
private IARecordBuilder builder;
- public ObjectConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public ObjectConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
public ObjectConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, GroupType parquetType,
- ParserContext context) {
- super(parent, fieldName, index, parquetType, context);
+ ParserContext context, List<Warning> warnings) {
+ super(parent, fieldName, index, parquetType, context, warnings);
}
@Override
@@ -59,7 +65,7 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
try {
builder.addField(value.getFieldName(), getValue());
} catch (HyracksDataException e) {
@@ -68,31 +74,34 @@
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
try {
- return new AtomicConverter(this, context.getSerializedFieldName(type.getFieldName(index)), index, context);
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ IValueReference fieldName = context.getSerializedFieldName(type.getFieldName(index));
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, fieldName, index, context,
+ warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
try {
final IValueReference childFieldName = context.getSerializedFieldName(type.getFieldName(index));
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, childFieldName, index, arrayType, context);
+ return new ArrayConverter(this, childFieldName, index, arrayType, context, warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
try {
final IValueReference childFieldName = context.getSerializedFieldName(type.getFieldName(index));
final GroupType objectType = type.getType(index).asGroupType();
- return new ObjectConverter(this, childFieldName, index, objectType, context);
+ return new ObjectConverter(this, childFieldName, index, objectType, context, warnings);
} catch (IOException e) {
throw new IllegalStateException(e);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
index b06afd3..3e4d811 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
@@ -51,10 +51,8 @@
@Override
public void close() throws IOException {
super.close();
- if (warningCollector.shouldWarn()) {
- //report warnings
- HDFSUtils.issueWarnings(warningCollector, conf);
- }
+ //report warnings
+ HDFSUtils.issueWarnings(warningCollector, conf);
}
@SuppressWarnings("unchecked")
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
index b2a5eeb..2f219e0 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
@@ -19,11 +19,10 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.IOException;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.HDFSUtils;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.runtime.projection.FunctionCallInformation;
@@ -40,20 +39,19 @@
@Override
public ReadContext init(InitContext context) {
MessageType requestedSchema = getRequestedSchema(context);
- return new ReadContext(requestedSchema, Collections.emptyMap());
+ return new ReadContext(requestedSchema);
}
@Override
public RecordMaterializer<IValueReference> prepareForRead(Configuration configuration,
Map<String, String> keyValueMetaData, MessageType fileSchema, ReadContext readContext) {
- return new ADMRecordMaterializer(readContext);
+ return new ADMRecordMaterializer(configuration, readContext);
}
private static MessageType getRequestedSchema(InitContext context) {
Configuration configuration = context.getConfiguration();
MessageType fileSchema = context.getFileSchema();
- boolean shouldWarn = configuration.getBoolean(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_ENABLED, false);
- AsterixTypeToParquetTypeVisitor visitor = new AsterixTypeToParquetTypeVisitor(shouldWarn);
+ AsterixTypeToParquetTypeVisitor visitor = new AsterixTypeToParquetTypeVisitor();
try {
ARecordType expectedType = HDFSUtils.getExpectedType(configuration);
Map<String, FunctionCallInformation> functionCallInformationMap =
@@ -61,7 +59,7 @@
MessageType requestedType = visitor.clipType(expectedType, fileSchema, functionCallInformationMap);
List<Warning> warnings = visitor.getWarnings();
- if (shouldWarn && !warnings.isEmpty()) {
+ if (!warnings.isEmpty()) {
//New warnings were created, set the warnings in hadoop configuration to be reported
HDFSUtils.setWarnings(warnings, configuration);
//Update the reported warnings so that we do not report the same warning again
@@ -76,8 +74,17 @@
private static class ADMRecordMaterializer extends RecordMaterializer<IValueReference> {
private final RootConverter rootConverter;
- public ADMRecordMaterializer(ReadContext readContext) {
- rootConverter = new RootConverter(readContext.getRequestedSchema());
+ public ADMRecordMaterializer(Configuration configuration, ReadContext readContext) {
+ List<Warning> warnings = new ArrayList<>();
+ rootConverter = new RootConverter(readContext.getRequestedSchema(), warnings);
+ try {
+ if (!warnings.isEmpty()) {
+ //Issue warnings if encountered an unsupported type
+ HDFSUtils.setWarnings(warnings, configuration);
+ }
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
}
@Override
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
index d46d84c..b249f8b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RepeatedConverter.java
@@ -19,13 +19,19 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve.PrimitiveConverterProvider;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
+import org.apache.parquet.io.api.PrimitiveConverter;
import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.PrimitiveType;
class RepeatedConverter extends AbstractComplexConverter {
- public RepeatedConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context) {
- super(parent, index, parquetType, context);
+ public RepeatedConverter(AbstractComplexConverter parent, int index, GroupType parquetType, ParserContext context,
+ List<Warning> warnings) {
+ super(parent, index, parquetType, context, warnings);
}
@Override
@@ -39,28 +45,29 @@
}
@Override
- protected void addValue(IFieldValue value) {
+ public void addValue(IFieldValue value) {
parent.addValue(value);
}
@Override
- protected AtomicConverter createAtomicConverter(GroupType type, int index) {
- return new AtomicConverter(this, index, context);
+ protected PrimitiveConverter createAtomicConverter(GroupType type, int index, List<Warning> warnings) {
+ PrimitiveType primitiveType = type.getType(index).asPrimitiveType();
+ return PrimitiveConverterProvider.createPrimitiveConverter(primitiveType, this, index, context, warnings);
}
@Override
- protected ArrayConverter createArrayConverter(GroupType type, int index) {
+ protected ArrayConverter createArrayConverter(GroupType type, int index, List<Warning> warnings) {
final GroupType arrayType = type.getType(index).asGroupType();
- return new ArrayConverter(this, index, arrayType, context);
+ return new ArrayConverter(this, index, arrayType, context, warnings);
}
@Override
- protected ObjectConverter createObjectConverter(GroupType type, int index) {
- return new ObjectConverter(this, index, type.getType(index).asGroupType(), context);
+ protected ObjectConverter createObjectConverter(GroupType type, int index, List<Warning> warnings) {
+ return new ObjectConverter(this, index, type.getType(index).asGroupType(), context, warnings);
}
@Override
- protected DataOutput getDataOutput() {
+ public DataOutput getDataOutput() {
return getParentDataOutput();
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
index 76f4342..d7f8018 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/RootConverter.java
@@ -19,8 +19,10 @@
package org.apache.asterix.external.input.record.reader.hdfs.parquet;
import java.io.DataOutput;
+import java.util.List;
import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.data.std.api.IValueReference;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.parquet.schema.GroupType;
@@ -28,8 +30,8 @@
class RootConverter extends ObjectConverter {
private final ArrayBackedValueStorage rootBuffer;
- public RootConverter(GroupType parquetType) {
- super(null, -1, parquetType, new ParserContext(true));
+ public RootConverter(GroupType parquetType, List<Warning> warnings) {
+ super(null, -1, parquetType, new ParserContext(true), warnings);
this.rootBuffer = new ArrayBackedValueStorage();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
new file mode 100644
index 0000000..9506e07
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/BinaryConverter.java
@@ -0,0 +1,59 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder;
+import org.apache.parquet.io.api.Binary;
+
+public class BinaryConverter extends GenericPrimitiveConverter {
+ private byte[] lengthBytes;
+
+ BinaryConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ @Override
+ public void addBinary(Binary value) {
+ DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.BINARY.serialize());
+ writeLength(value.length(), out);
+ value.writeTo(out);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ private void writeLength(int length, DataOutput out) throws IOException {
+ int requiredLength = VarLenIntEncoderDecoder.getBytesRequired(length);
+ if (lengthBytes == null || requiredLength > lengthBytes.length) {
+ lengthBytes = new byte[requiredLength];
+ }
+ VarLenIntEncoderDecoder.encode(length, lengthBytes, 0);
+ out.write(lengthBytes, 0, requiredLength);
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
new file mode 100644
index 0000000..f33865c
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/DateConverter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+
+class DateConverter extends GenericPrimitiveConverter {
+ DateConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ @Override
+ public void addInt(int value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATE.serialize());
+ out.writeInt(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
similarity index 83%
rename from asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java
rename to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
index 7219bdd..88b65ef 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/GenericPrimitiveConverter.java
@@ -16,11 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+package org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve;
import java.io.DataOutput;
import java.io.IOException;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.IFieldValue;
import org.apache.asterix.external.input.stream.StandardUTF8ToModifiedUTF8DataOutput;
import org.apache.asterix.external.parser.jackson.ParserContext;
import org.apache.asterix.om.types.ATypeTag;
@@ -28,20 +30,14 @@
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.io.api.PrimitiveConverter;
-/**
- * Currently, only JSON types are supported (string, number, boolean)
- */
-class AtomicConverter extends PrimitiveConverter implements IFieldValue {
- private final AbstractComplexConverter parent;
- private final IValueReference fieldName;
- private final int index;
- private final ParserContext context;
+public class GenericPrimitiveConverter extends PrimitiveConverter implements IFieldValue {
- public AtomicConverter(AbstractComplexConverter parent, int index, ParserContext context) {
- this(parent, null, index, context);
- }
+ protected final AbstractComplexConverter parent;
+ protected final IValueReference fieldName;
+ protected final int index;
+ protected final ParserContext context;
- public AtomicConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
+ GenericPrimitiveConverter(AbstractComplexConverter parent, IValueReference fieldName, int index,
ParserContext context) {
this.parent = parent;
this.fieldName = fieldName;
@@ -50,6 +46,16 @@
}
@Override
+ public final IValueReference getFieldName() {
+ return fieldName;
+ }
+
+ @Override
+ public final int getIndex() {
+ return index;
+ }
+
+ @Override
public void addBinary(Binary value) {
final DataOutput out = parent.getDataOutput();
final StandardUTF8ToModifiedUTF8DataOutput stringOut = context.getModifiedUTF8DataOutput();
@@ -108,14 +114,4 @@
}
parent.addValue(this);
}
-
- @Override
- public IValueReference getFieldName() {
- return fieldName;
- }
-
- @Override
- public int getIndex() {
- return index;
- }
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
similarity index 98%
rename from asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java
rename to asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
index e38056b..21b9217 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/MissingConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/MissingConverter.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.asterix.external.input.record.reader.hdfs.parquet;
+package org.apache.asterix.external.input.record.reader.hdfs.parquet.primitve;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.io.api.PrimitiveConverter;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
new file mode 100644
index 0000000..9fe2b42
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/PrimitiveConverterProvider.java
@@ -0,0 +1,75 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.util.List;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AsterixTypeToParquetTypeVisitor;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.api.exceptions.Warning;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.PrimitiveConverter;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Types;
+
+public class PrimitiveConverterProvider {
+ public static final PrimitiveType MISSING =
+ Types.optional(PrimitiveType.PrimitiveTypeName.BOOLEAN).named("MISSING");
+
+ private PrimitiveConverterProvider() {
+ }
+
+ public static PrimitiveConverter createPrimitiveConverter(PrimitiveType type, AbstractComplexConverter parent,
+ int index, ParserContext context, List<Warning> warnings) {
+ return createPrimitiveConverter(type, parent, null, index, context, warnings);
+ }
+
+ public static PrimitiveConverter createPrimitiveConverter(PrimitiveType type, AbstractComplexConverter parent,
+ IValueReference fieldName, int index, ParserContext context, List<Warning> warnings) {
+
+ if (type == MISSING) {
+ return MissingConverter.INSTANCE;
+ }
+
+ ATypeTag mappedType = AsterixTypeToParquetTypeVisitor.mapType(type);
+ switch (mappedType) {
+ case BOOLEAN:
+ case BIGINT:
+ case DOUBLE:
+ case STRING:
+ return new GenericPrimitiveConverter(parent, fieldName, index, context);
+ case BINARY:
+ return new BinaryConverter(parent, fieldName, index, context);
+ case DATE:
+ return new DateConverter(parent, fieldName, index, context);
+ case DATETIME:
+ return new TimestampConverter(parent, fieldName, index, context);
+ default:
+ return warn(warnings, type);
+ }
+ }
+
+ private static PrimitiveConverter warn(List<Warning> warnings, PrimitiveType type) {
+ warnings.add(Warning.of(null, ErrorCode.UNSUPPORTED_PARQUET_TYPE, type.toString()));
+ return MissingConverter.INSTANCE;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
new file mode 100644
index 0000000..6e6fa09
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/primitve/TimestampConverter.java
@@ -0,0 +1,96 @@
+/*
+ * 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.asterix.external.input.record.reader.hdfs.parquet.primitve;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import org.apache.asterix.external.input.record.reader.hdfs.parquet.AbstractComplexConverter;
+import org.apache.asterix.external.parser.jackson.ParserContext;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.parquet.io.api.Binary;
+
+class TimestampConverter extends GenericPrimitiveConverter {
+ private static final long JULIAN_DAY_OF_EPOCH = 2440588;
+ private static final long MILLIS_PER_DAY = 86400000L;
+ private static final long NANOS_PER_MILLIS = 1000000L;
+
+ TimestampConverter(AbstractComplexConverter parent, IValueReference fieldName, int index, ParserContext context) {
+ super(parent, fieldName, index, context);
+ }
+
+ /**
+ * Timestamp is an INT96 (Little Endian)
+ * Skip nanoseconds part
+ *
+ * @param value binary representation of INT96
+ */
+ @Override
+ public void addBinary(Binary value) {
+ DataOutput out = parent.getDataOutput();
+ ByteBuffer buffer = value.toByteBuffer().order(ByteOrder.LITTLE_ENDIAN);
+ long timeOfDayNanos = buffer.getLong();
+ int julianDay = buffer.getInt();
+ long timestamp = fromJulian(julianDay, timeOfDayNanos);
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(timestamp);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+
+ parent.addValue(this);
+ }
+
+ @Override
+ public void addInt(int value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ /**
+ * Timestamp is an INT64
+ *
+ * @param value long value
+ */
+ @Override
+ public void addLong(long value) {
+ final DataOutput out = parent.getDataOutput();
+ try {
+ out.writeByte(ATypeTag.DATETIME.serialize());
+ out.writeLong(value);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ parent.addValue(this);
+ }
+
+ private static long fromJulian(int days, long nanos) {
+ return (days - JULIAN_DAY_OF_EPOCH) * MILLIS_PER_DAY + nanos / NANOS_PER_MILLIS;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index ec25997..16a0f66 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -60,7 +60,6 @@
public static final String KEY_HADOOP_SHORT_CIRCUIT = "dfs.client.read.shortcircuit";
public static final String KEY_HADOOP_SOCKET_PATH = "dfs.domain.socket.path";
public static final String KEY_HADOOP_BUFFER_SIZE = "io.file.buffer.size";
- public static final String KEY_HADOOP_ASTERIX_WARNINGS_ENABLED = "org.apache.asterix.warnings.enabled";
//Base64 encoded warnings issued from Hadoop
public static final String KEY_HADOOP_ASTERIX_WARNINGS_LIST = "org.apache.asterix.warnings.list";
//Disable caching FileSystem for Hadoop
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index 9f65cd7..25cd832 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -201,7 +201,7 @@
}
}
- public static JobConf configureHDFSJobConf(Map<String, String> configuration, boolean shouldWarn) {
+ public static JobConf configureHDFSJobConf(Map<String, String> configuration) {
JobConf conf = new JobConf();
String localShortCircuitSocketPath = configuration.get(ExternalDataConstants.KEY_LOCAL_SOCKET_PATH);
String formatClassName = HDFSUtils.getInputFormatClassName(configuration);
@@ -235,11 +235,6 @@
//Subset of the values were requested, set the functionCallInformation
conf.set(ExternalDataConstants.KEY_HADOOP_ASTERIX_FUNCTION_CALL_INFORMATION,
configuration.get(ExternalDataConstants.KEY_HADOOP_ASTERIX_FUNCTION_CALL_INFORMATION));
- /*
- * Allows Parquet to issue warnings in case we found type mismatches (if warnings are enabled).
- * Warnings will be issued during the type matching of Parquet's schema with the requested schema
- */
- conf.setBoolean(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_ENABLED, shouldWarn);
}
conf.set(ExternalDataConstants.KEY_REQUESTED_FIELDS, requestedValues);
}
@@ -320,10 +315,10 @@
public static void issueWarnings(IWarningCollector warningCollector, Configuration conf) throws IOException {
String warnings = conf.get(ExternalDataConstants.KEY_HADOOP_ASTERIX_WARNINGS_LIST, "");
- if (!warnings.isEmpty() && warningCollector.shouldWarn()) {
+ if (!warnings.isEmpty()) {
String[] encodedWarnings = warnings.split(",");
Base64.Decoder decoder = Base64.getDecoder();
- for (int i = 0; i < encodedWarnings.length; i++) {
+ for (int i = 0; i < encodedWarnings.length && warningCollector.shouldWarn(); i++) {
/*
* This should create a small number of objects as warnings are reported only once by AsterixDB's
* hadoop readers
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
index 5cb26fd..ee2f168 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/projection/FunctionCallInformation.java
@@ -73,6 +73,10 @@
ExceptionUtil.indexToPosition(0), expectedType, actualType);
}
+ public Warning createWarning(ErrorCode errorCode, Serializable... args) {
+ return Warning.of(sourceLocation, errorCode, args);
+ }
+
public void writeFields(DataOutput output) throws IOException {
output.writeUTF(functionName);
SourceLocation.writeFields(sourceLocation, output);
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 1
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newchange
Change in asterixdb[neo]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Wail Alkowaileet <wa...@gmail.com>:
Wail Alkowaileet has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 10: Code-Review+1
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 10
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Comment-Date: Fri, 28 Jan 2022 15:44:17 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
Anon. E. Moose #1000171 has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 5:
Analytics Compatibility Tests Failed
https://cbjenkins.page.link/xMZqPd9rEecFY1pE9 : UNSTABLE
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 5
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Comment-Date: Thu, 13 Jan 2022 21:54:27 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: No
Gerrit-MessageType: comment
Change in asterixdb[neo]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Wail Alkowaileet <wa...@gmail.com>:
Hello Dmitry Lychagin, Ali Alsuliman, Jenkins, Anon. E. Moose #1000171,
I'd like you to reexamine a change. Please visit
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
to look at the new patch set (#6).
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
[ASTERIXDB-2981][EXT] Handle Parquet specialized types
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Handle parquet specialized types
- Supported types: Date, Timestamp, Binary, JSON,
BSON (as Binary), and Decimal (converted to double)
- By default, type conversions are not enabled
- To enable type conversions, options must be passed
during dataset creation (see parquet-types.1.ddl.sqlpp)
Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
---
A asterixdb/asterix-app/data/hdfs/parquet/binary/parquetTypes.parquet
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/BinaryFileConverterUtil.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.1.ddl.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.2.query.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/parquet-types.3.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.02.adm
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/parquet-types.03.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
M asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
A asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/IFieldValue.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/AbstractComplexConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/ArrayConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/ObjectConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RepeatedConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/BinaryConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/DateConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/DecimalConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/GenericPrimitiveConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/JsonStringConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/MissingConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/PrimitiveConverterProvider.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/TimestampConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/jackson/ParserContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
32 files changed, 1,220 insertions(+), 174 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/03/13903/6
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 6
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newpatchset
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
Anon. E. Moose #1000171 has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 1:
Analytics Compatibility Compilation Successful
https://cbjenkins.page.link/PRj3kuXsWouQfmiLA : SUCCESS
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 1
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-CC: Anon. E. Moose #1000171
Gerrit-CC: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Comment-Date: Mon, 01 Nov 2021 23:21:37 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: No
Gerrit-MessageType: comment
Change in asterixdb[master]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Jenkins <je...@fulliautomatix.ics.uci.edu>:
Jenkins has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 1: Integration-Tests+1
Integration Tests Successful
https://asterix-jenkins.ics.uci.edu/job/asterix-gerrit-integration-tests/12668/ : SUCCESS
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 1
Gerrit-Owner: Wael Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-CC: Anon. E. Moose #1000171
Gerrit-Comment-Date: Tue, 02 Nov 2021 00:33:45 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
Change in asterixdb[neo]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Dmitry Lychagin <dm...@couchbase.com>:
Dmitry Lychagin has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 10:
(17 comments)
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md
File asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md:
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@41
PS10, Line 41: CREATE TYPE ParquetType AS {
Future thought. I think we should not require user to create this empty type for parquet datasets. We should make DatasetTypeSpecification() optional in ExternalDatasetSpecification() and assume that if it's not provided then the the type is MetadataBuiltinEntities.ANY_OBJECT_DATATYPE
See https://github.com/apache/asterixdb/blob/master/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj#L1069
If we do that then user will be able to just declare external datasets as follows:
CREATE EXTERNAL DATASET ParquetDataset USING S3 ....
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@110
PS10, Line 110: AsterixDB supports Parquet's generic types such String, Int and Double. However, Parquet files could
Let's use upper case type names here: STRING, INT, DOUBLE
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@112
PS10, Line 112: DATETIME like types. The following table show the type mapping between Apache Parquet and AsterixDB:
What about Parquet's LIST/MAP datatypes? I don't see them documented here.
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@225
PS10, Line 225: <td>Time in milliseconds. The time could be adjusted to UTC.
"could be adjusted to UTC" is not clear. Who's doing the adjustment? AsterixDB? I think we should say that this type was stored in Parquet with isAdjustedToUTC = true option then we'll read it as local UTC time. Also, I think timeZoneId should not apply in this case (for TIME datatype). If timeZoneId is given as a named timezone (America/New_York) then which date are we going to use to compute UTC offset?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@233
PS10, Line 233: <td>Time in micro/nano seconds. The time could be adjusted to UTC.
same comment as for INT_32(TIME)
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@240
PS10, Line 240: <td><code>DATETIME</code></td>
is INT96 always supposed to be mapped to DATETIME? Are there any cases where user might want to map it to DOUBLE instead?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@246
PS10, Line 246: <td><code>ANY</code></td>
I think we don't mention ANY type explicitly in the docs. May be just say "any type" instead.
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@268
PS10, Line 268: <td>N/A</td>
Can you clarify here. What happens if BSON type is encountered? An error or MISSING with warning?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@274
PS10, Line 274: <td>Not supported.</td>
Need to clarify what happens if it's encountered.
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@283
PS10, Line 283: of a native support for `DECIMAL` type in AsterixDB.
Can we remove this statement? "For example, the introduction
of a native support for `DECIMAL` type in AsterixDB."
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@295
PS10, Line 295: ("decimalToDouble" = "true")
We currently mostly use "-" as word separator in ExternalDataConstants. Should we rename this flag to "decimal-to-double" to be consistent?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@298
PS10, Line 298: This flag will enable parsing/converting `DECIMAL` values/types into a proper AsterixDB value/type. When the flag is not
I'd say "DOUBLE" instead of "proper" here. so ".. into a DOUBLE AsterixDB value/type"
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@311
PS10, Line 311: provide the timezone ID to adjust their values to the preferred local value by setting the flag `timeZoneId`. To do so,
As I mentioned before, I think time zone flag should only apply to "DATETIME" values, not "TIME" values.
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@320
PS10, Line 320: ("timeZoneId" = "PST")
can we rename it to just "timezone"?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@323
PS10, Line 323: If the flag `timeZoneId` is not set (or it is invalid), a warning will appear when running a query:
I think if the timezone value is invalid then we should fail at dataset creation time, and also detect this and fail at query runtime.
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@341
PS10, Line 341: ("parseJsonString" = "false")
use "-" as word separators?
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903/10/asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md@348
PS10, Line 348: Currently, AsterixDB do not support Parquet's `INTERVAL` type.
Explain what happens if it's encountered.
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 10
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Comment-Date: Fri, 28 Jan 2022 19:28:01 +0000
Gerrit-HasComments: Yes
Gerrit-Has-Labels: No
Gerrit-MessageType: comment
Change in asterixdb[neo]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Wail Alkowaileet <wa...@gmail.com>:
Hello Dmitry Lychagin, Ali Alsuliman, Jenkins, Anon. E. Moose #1000171,
I'd like you to reexamine a change. Please visit
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
to look at the new patch set (#12).
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
[ASTERIXDB-2981][EXT] Handle Parquet specialized types
- user model changes: yes
- storage format changes: no
- interface changes: no
Details:
- Handle parquet specialized types
- Supported types: Date, Time, Timestamp, Binary, JSON,
BSON (as Binary), Decimal (converted to double), UUID
- By default, type conversion for decimal is not enabled
- To enable decimal convertion, an option must be passed
during dataset creation (see parquet-types.1.ddl.sqlpp)
- UTC-adjusted values can readjusted to local values by
setting the the timezone ID using "timeZoneId" option
upon creating a dataset.
- Documentation is also included in this patch
Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
---
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/aws/AwsS3ExternalDatasetTest.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/microsoft/AzureBlobStorageExternalDatasetTest.java
R asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/parquet/BinaryFileConverterUtil.java
A asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/parquet/ParquetFileExampleGeneratorUtil.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/HDFSCluster.java
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/decimal/decimal.1.ddl.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/decimal/decimal.2.query.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/decimal/decimal.3.query.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/disable-json-parsing/disable-json-parsing.1.ddl.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/disable-json-parsing/disable-json-parsing.2.query.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/invalid-timezone/temporal.1.ddl.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/temporal/temporal.1.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/temporal/tempral.2.query.sqlpp
C asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/unset-flags/unset-flags.1.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/parquet/parquet-types/unset-flags/unset-flags.2.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/decimal/decimal.02.adm
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/disable-json-parsing/disable-json-parsing.02.adm
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/temporal/temporal.02.adm
A asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/parquet/parquet-types/unset-flags/unset-flags.02.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
M asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
A asterixdb/asterix-doc/src/site/markdown/sqlpp/parquet.md
C asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixParquetRuntimeException.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AsterixTypeToParquetTypeVisitor.java
D asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/AtomicConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetFileRecordReader.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/ParquetReadSupport.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/IFieldValue.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/ParquetConverterContext.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/AbstractComplexConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/ArrayConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/ObjectConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RepeatedConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/nested/RootConverter.java
C asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/BinaryConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/DateConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/DecimalConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/GenericPrimitiveConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/JsonStringConverter.java
R asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/MissingConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/PrimitiveConverterProvider.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/TimeConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/TimestampConverter.java
C asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/UUIDConverter.java
A asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/hdfs/parquet/converter/primitve/UnsignedIntegerConverter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/parser/jackson/ParserContext.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
52 files changed, 2,136 insertions(+), 438 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/03/13903/12
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 12
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-MessageType: newpatchset
Change in asterixdb[neo]: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
Posted by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu>.
From Jenkins <je...@fulliautomatix.ics.uci.edu>:
Jenkins has posted comments on this change. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903 )
Change subject: [ASTERIXDB-2981][EXT] Handle Parquet specialized types
......................................................................
Patch Set 13: Integration-Tests+1
Integration Tests Successful
https://asterix-jenkins.ics.uci.edu/job/asterix-gerrit-integration-tests/12931/ : SUCCESS
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/13903
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: neo
Gerrit-Change-Id: I345809e16aeb91f0a28d6592e1b8799e48434051
Gerrit-Change-Number: 13903
Gerrit-PatchSet: 13
Gerrit-Owner: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Reviewer: Ali Alsuliman <al...@gmail.com>
Gerrit-Reviewer: Anon. E. Moose #1000171
Gerrit-Reviewer: Dmitry Lychagin <dm...@couchbase.com>
Gerrit-Reviewer: Jenkins <je...@fulliautomatix.ics.uci.edu>
Gerrit-Reviewer: Wail Alkowaileet <wa...@gmail.com>
Gerrit-Comment-Date: Mon, 31 Jan 2022 21:25:08 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment