>From <[email protected]>: [email protected] has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/19773 )
Change subject: [ASTERIXDB-3392] Handle NULL/Empty types in Copy to parquet ...................................................................... [ASTERIXDB-3392] Handle NULL/Empty types in Copy to parquet Change-Id: I53a54b725e360f586f5e8e83f1692d77f0a5e93f --- M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/SchemaCheckerLazyVisitor.java M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaTree.java M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.04.ddl.sqlpp M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSchemaInferPoolWriter.java M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetRecordLazyVisitor.java A asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/parquet-null-type/parquet-null-type.05.adm A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.01.ddl.sqlpp A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.05.query.sqlpp M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaLazyVisitor.java A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.02.update.sqlpp A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.03.update.sqlpp 12 files changed, 244 insertions(+), 16 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/73/19773/1 diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.01.ddl.sqlpp new file mode 100644 index 0000000..8f9bb53 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.01.ddl.sqlpp @@ -0,0 +1,29 @@ +/* + * 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. + */ + +DROP DATAVERSE test if exists; +CREATE DATAVERSE test; +USE test; + + +CREATE TYPE ColumnType1 AS { + id: integer + }; + +CREATE COLLECTION TestCollection(ColumnType1) PRIMARY KEY id; \ No newline at end of file diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.02.update.sqlpp new file mode 100644 index 0000000..8f50900 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.02.update.sqlpp @@ -0,0 +1,27 @@ +/* + * 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. + */ + +use test; + + + +insert into TestCollection({"id":8,"name":null, "nested" : { "List":[100,null,300] } }); +insert into TestCollection({"id":10,"name":"Virat", "nested" : { "List":[] } }); +insert into TestCollection({"id":28,"name":"Virat", "nested" : { "List":null , "A" : "a" } }); +-- insert into TestCollection({"id":30, "nested" : { "list":[300, 400 ] , "randomK" : "randomV" } }); diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.03.update.sqlpp new file mode 100644 index 0000000..fa70214 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.03.update.sqlpp @@ -0,0 +1,34 @@ +/* + * 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. + */ + +USE test; + +COPY ( +select c.* from TestCollection c + ) toWriter +TO %adapter% +PATH (%pathprefix% "copy-to-result", "parquet-null-type") +WITH { + %template_colons%, + %additionalProperties% + "format":"parquet" + }; + + + diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.04.ddl.sqlpp new file mode 100644 index 0000000..0773745 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.04.ddl.sqlpp @@ -0,0 +1,36 @@ +/* + * 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. + */ + +USE test; + + +CREATE TYPE ColumnType2 AS { + }; + + + +CREATE EXTERNAL DATASET TestDataset(ColumnType2) USING %adapter% +( + %template%, + %additional_Properties%, + ("definition"="%path_prefix%copy-to-result/parquet-null-type/"), + ("include"="*.parquet"), + ("requireVersionChangeDetection"="false"), + ("format" = "parquet") +); diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.05.query.sqlpp new file mode 100644 index 0000000..b03fc5e --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/parquet-null-type/parquet-null-type.05.query.sqlpp @@ -0,0 +1,27 @@ +/* + * 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. + */ + +USE test; + + +SELECT c.* +FROM TestDataset c +ORDER BY c.id; + + diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/parquet-null-type/parquet-null-type.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/parquet-null-type/parquet-null-type.05.adm new file mode 100644 index 0000000..6c88565 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/parquet-null-type/parquet-null-type.05.adm @@ -0,0 +1,3 @@ +{ "id": 8, "nested": { "List": [ 100, 300 ] } } +{ "name": "Virat", "id": 10, "nested": { "List": [ ] } } +{ "name": "Virat", "id": 28, "nested": { "A": "a" } } 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 c9f9da6..fa4b7dd 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 @@ -115,6 +115,16 @@ </compilation-unit> </test-case> <test-case FilePath="copy-to"> + <compilation-unit name="parquet-null-type"> + <placeholder name="adapter" value="S3" /> + <placeholder name="pathprefix" value="" /> + <placeholder name="path_prefix" value="" /> + <placeholder name="additionalProperties" value='"container":"playground",' /> + <placeholder name="additional_Properties" value='("container"="playground")' /> + <output-dir compare="Text">parquet-null-type</output-dir> + </compilation-unit> + </test-case> + <test-case FilePath="copy-to"> <compilation-unit name="parquet-file-writers"> <placeholder name="adapter" value="S3" /> <placeholder name="pathprefix" value="" /> diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSchemaInferPoolWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSchemaInferPoolWriter.java index ca87ced..b0c567b 100644 --- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSchemaInferPoolWriter.java +++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSchemaInferPoolWriter.java @@ -62,7 +62,7 @@ return; } else if (schemaComparisonType.equals(ISchemaChecker.SchemaComparisonType.GROWING)) { // If the schema is growing, close the existing writer and create a new one with the new schema. - schemaNodes.set(i, schemaLazyVisitor.inferSchema(value)); + schemaNodes.set(i, schemaLazyVisitor.inferSchema(value, schemaNodes.get(i))); closeWriter(i); return; } @@ -75,14 +75,15 @@ }); throw new HyracksDataException(ErrorCode.SCHEMA_LIMIT_EXCEEDED, maxSchemas); } - schemaNodes.add(schemaLazyVisitor.inferSchema(value)); + schemaNodes.add(schemaLazyVisitor.inferSchema(value, null)); writerList.add(null); } public void write(IValueReference value, IFrameTupleReference tupleRef) throws HyracksDataException { for (int i = 0; i < schemaNodes.size(); i++) { - if (schemaChecker.checkSchema(schemaNodes.get(i), value) - .equals(ISchemaChecker.SchemaComparisonType.EQUIVALENT)) { + ISchemaChecker.SchemaComparisonType schemaComparisonType = + schemaChecker.checkSchema(schemaNodes.get(i), value); + if (schemaComparisonType.equals(ISchemaChecker.SchemaComparisonType.EQUIVALENT)) { createOrWrite(i, value, tupleRef); return; } diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetRecordLazyVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetRecordLazyVisitor.java index cffeb2f..79367c8 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetRecordLazyVisitor.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetRecordLazyVisitor.java @@ -85,6 +85,11 @@ for (int i = 0; i < pointable.getNumberOfChildren(); i++) { pointable.nextChild(); AbstractLazyVisitablePointable child = pointable.getChildVisitablePointable(); + + if (child.getTypeTag() == ATypeTag.NULL || child.getTypeTag() == ATypeTag.MISSING) { + continue; + } + String columnName = fieldNamesDictionary.getOrCreateFieldNameIndex(pointable.getFieldName()); if (!groupType.containsField(columnName)) { @@ -144,6 +149,12 @@ AbstractLazyVisitablePointable child = pointable.getChildVisitablePointable(); recordConsumer.startGroup(); + + if (child.getTypeTag() == ATypeTag.NULL) { + recordConsumer.endGroup(); + continue; + } + recordConsumer.startField(ELEMENT_FIELD, listType.getFieldIndex(ELEMENT_FIELD)); child.accept(this, listType.getType(ELEMENT_FIELD)); recordConsumer.endField(ELEMENT_FIELD, listType.getFieldIndex(ELEMENT_FIELD)); @@ -166,6 +177,7 @@ throw new HyracksDataException(ErrorCode.RESULT_DOES_NOT_FOLLOW_SCHEMA, PRIMITIVE_TYPE_ERROR_FIELD, GROUP_TYPE_ERROR_FIELD, type.getName()); } + parquetValueWriter.addValueToColumn(recordConsumer, pointable, type.asPrimitiveType()); return null; } @@ -181,6 +193,10 @@ String columnName = fieldNamesDictionary.getOrCreateFieldNameIndex(rec.getFieldName()); AbstractLazyVisitablePointable child = rec.getChildVisitablePointable(); + if (child.getTypeTag() == ATypeTag.NULL) { + continue; + } + if (!schema.containsField(columnName)) { LOGGER.info("Schema: {} does not contain field: {}", LogRedactionUtil.userData(schema.toString()), LogRedactionUtil.userData(columnName)); diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaLazyVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaLazyVisitor.java index 2278372..4a07695 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaLazyVisitor.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaLazyVisitor.java @@ -66,15 +66,19 @@ if (schemaNode.getType() == null) { schemaNode.setType(new ParquetSchemaTree.RecordType()); } - if (!(schemaNode.getType() instanceof ParquetSchemaTree.RecordType)) { + if (!(schemaNode.getType() instanceof ParquetSchemaTree.RecordType recordType)) { LOGGER.info("Incompatible type found in record: {} and {}", LogRedactionUtil.userData(schemaNode.toString()), pointable.getTypeTag()); throw RuntimeDataException.create(PARQUET_UNSUPPORTED_MIXED_TYPE_ARRAY); } - ParquetSchemaTree.RecordType recordType = (ParquetSchemaTree.RecordType) schemaNode.getType(); for (int i = 0; i < pointable.getNumberOfChildren(); i++) { pointable.nextChild(); AbstractLazyVisitablePointable child = pointable.getChildVisitablePointable(); + + if (child.getTypeTag() == ATypeTag.NULL) { + continue; + } + String childColumnName = fieldNamesDictionary.getOrCreateFieldNameIndex(pointable.getFieldName()); ParquetSchemaTree.SchemaNode childType; if (recordType.getChildren().containsKey(childColumnName)) { @@ -102,6 +106,11 @@ for (int i = 0; i < numChildren; i++) { pointable.nextChild(); AbstractLazyVisitablePointable child = pointable.getChildVisitablePointable(); + + if (child.getTypeTag() == ATypeTag.NULL) { + continue; + } + if (listType.isEmpty()) { listType.setChild(new ParquetSchemaTree.SchemaNode()); } @@ -113,19 +122,21 @@ @Override public Void visit(FlatLazyVisitablePointable pointable, ParquetSchemaTree.SchemaNode schemaNode) throws HyracksDataException { + + + if (schemaNode.getType() == null) { - if (!AsterixParquetTypeMap.PRIMITIVE_TYPE_NAME_MAP.containsKey(pointable.getTypeTag())) { - throw RuntimeDataException.create(TYPE_UNSUPPORTED_PARQUET_WRITE, pointable.getTypeTag()); + if (pointable.getTypeTag() == ATypeTag.NULL || AsterixParquetTypeMap.PRIMITIVE_TYPE_NAME_MAP.containsKey(pointable.getTypeTag()) ) { + schemaNode.setType(new ParquetSchemaTree.FlatType(pointable.getTypeTag())); + return null; } - schemaNode.setType(new ParquetSchemaTree.FlatType(pointable.getTypeTag())); - return null; + throw RuntimeDataException.create(TYPE_UNSUPPORTED_PARQUET_WRITE, pointable.getTypeTag()); } - if (!(schemaNode.getType() instanceof ParquetSchemaTree.FlatType)) { + if (!(schemaNode.getType() instanceof ParquetSchemaTree.FlatType flatType)) { LOGGER.info("Incompatible type found: {} and {}", LogRedactionUtil.userData(schemaNode.toString()), pointable.getTypeTag()); throw RuntimeDataException.create(PARQUET_UNSUPPORTED_MIXED_TYPE_ARRAY); } - ParquetSchemaTree.FlatType flatType = (ParquetSchemaTree.FlatType) schemaNode.getType(); if (!flatType.isCompatibleWith(pointable.getTypeTag())) { LOGGER.info("Incompatible type found: {} and {}", flatType, pointable.getTypeTag()); @@ -137,8 +148,14 @@ return null; } - public ParquetSchemaTree.SchemaNode inferSchema(IValueReference valueReference) throws HyracksDataException { - ParquetSchemaTree.SchemaNode schemaNode = new ParquetSchemaTree.SchemaNode(); + public ParquetSchemaTree.SchemaNode inferSchema(IValueReference valueReference, + ParquetSchemaTree.SchemaNode previousSchema) throws HyracksDataException { + ParquetSchemaTree.SchemaNode schemaNode; + if (previousSchema != null) { + schemaNode = previousSchema; + } else { + schemaNode = new ParquetSchemaTree.SchemaNode(); + } rec.set(valueReference); rec.accept(this, schemaNode); return schemaNode; diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaTree.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaTree.java index dae5295..70b20e9 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaTree.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/ParquetSchemaTree.java @@ -107,7 +107,14 @@ return typeTag.toString(); } + public boolean isNullType() { + return typeTag == ATypeTag.NULL; + } + public boolean isCompatibleWith(ATypeTag typeTag) { + if (isNullType() || typeTag == ATypeTag.NULL) { + return true; + } if (isHierarchical) { return AsterixParquetTypeMap.HIERARCHIAL_TYPES.containsKey(typeTag); } else { @@ -116,6 +123,11 @@ } public void coalesce(ATypeTag typeTag) { + if (isNullType()) { + this.typeTag = typeTag; + isHierarchical = AsterixParquetTypeMap.HIERARCHIAL_TYPES.containsKey(typeTag); + return; + } if (!isCompatibleWith(typeTag) || !isHierarchical) { return; } diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/SchemaCheckerLazyVisitor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/SchemaCheckerLazyVisitor.java index 28d4247..6d4495e 100644 --- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/SchemaCheckerLazyVisitor.java +++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/parquet/SchemaCheckerLazyVisitor.java @@ -109,11 +109,18 @@ if (schemaNode.getType() == null) { return ISchemaChecker.SchemaComparisonType.GROWING; } - if (!(schemaNode.getType() instanceof ParquetSchemaTree.FlatType)) { + + if(pointable.getTypeTag()==ATypeTag.NULL){ + return ISchemaChecker.SchemaComparisonType.EQUIVALENT; + } + + if (!(schemaNode.getType() instanceof ParquetSchemaTree.FlatType flatType)) { return ISchemaChecker.SchemaComparisonType.CONFLICTING; } - ParquetSchemaTree.FlatType flatType = (ParquetSchemaTree.FlatType) schemaNode.getType(); + if(flatType.isNullType() && pointable.getTypeTag()!= ATypeTag.NULL) { + return SchemaComparisonType.GROWING; + } if (!flatType.isCompatibleWith(pointable.getTypeTag())) { return ISchemaChecker.SchemaComparisonType.CONFLICTING; -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/19773 To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings Gerrit-Project: asterixdb Gerrit-Branch: ionic Gerrit-Change-Id: I53a54b725e360f586f5e8e83f1692d77f0a5e93f Gerrit-Change-Number: 19773 Gerrit-PatchSet: 1 Gerrit-Owner: [email protected] Gerrit-MessageType: newchange
