From commits-return-113206-archive-asf-public=cust-asf.ponee.io@beam.apache.org Fri Mar 5 08:57:30 2021 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mxout1-he-de.apache.org (mxout1-he-de.apache.org [95.216.194.37]) by mx-eu-01.ponee.io (Postfix) with ESMTPS id 5D8DE180645 for ; Fri, 5 Mar 2021 09:57:30 +0100 (CET) Received: from mail.apache.org (mailroute1-lw-us.apache.org [207.244.88.153]) by mxout1-he-de.apache.org (ASF Mail Server at mxout1-he-de.apache.org) with SMTP id BB09262DC2 for ; Fri, 5 Mar 2021 08:57:29 +0000 (UTC) Received: (qmail 54251 invoked by uid 500); 5 Mar 2021 08:57:28 -0000 Mailing-List: contact commits-help@beam.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@beam.apache.org Delivered-To: mailing list commits@beam.apache.org Received: (qmail 54242 invoked by uid 99); 5 Mar 2021 08:57:28 -0000 Received: from ec2-52-202-80-70.compute-1.amazonaws.com (HELO gitbox.apache.org) (52.202.80.70) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 05 Mar 2021 08:57:28 +0000 Received: by gitbox.apache.org (ASF Mail Server at gitbox.apache.org, from userid 33) id ACBD681760; Fri, 5 Mar 2021 08:57:28 +0000 (UTC) Date: Fri, 05 Mar 2021 08:57:23 +0000 To: "commits@beam.apache.org" Subject: [beam] branch master updated: [BEAM-7929] Support column projection for Parquet Tables MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit Message-ID: <161493463910.9304.7324518158224667550@gitbox.apache.org> From: iemejia@apache.org X-Git-Host: gitbox.apache.org X-Git-Repo: beam X-Git-Refname: refs/heads/master X-Git-Reftype: branch X-Git-Oldrev: d780ebd50af4e47faabd925082c1a324e15c1e4b X-Git-Newrev: d79cd82943c90dad518b705b7e81bcd2d2fc0f21 X-Git-Rev: d79cd82943c90dad518b705b7e81bcd2d2fc0f21 X-Git-NotificationType: ref_changed_plus_diff X-Git-Multimail-Version: 1.5.dev Auto-Submitted: auto-generated This is an automated email from the ASF dual-hosted git repository. iemejia pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git The following commit(s) were added to refs/heads/master by this push: new d79cd82 [BEAM-7929] Support column projection for Parquet Tables new 2447679 Merge pull request #14117: [BEAM-7929] Support column projection for Parquet Tables d79cd82 is described below commit d79cd82943c90dad518b705b7e81bcd2d2fc0f21 Author: Ismaël Mejía AuthorDate: Mon Mar 1 10:05:32 2021 +0100 [BEAM-7929] Support column projection for Parquet Tables --- sdks/java/extensions/sql/build.gradle | 1 + .../sql/meta/provider/parquet/ParquetTable.java | 132 +++++++++++++++++++++ .../provider/parquet/ParquetTableProvider.java | 22 ++-- .../provider/parquet/ParquetTableProviderTest.java | 35 +++++- .../sdk/io/parquet/ParquetSchemaIOProvider.java | 127 -------------------- 5 files changed, 172 insertions(+), 145 deletions(-) diff --git a/sdks/java/extensions/sql/build.gradle b/sdks/java/extensions/sql/build.gradle index 6de73f2..6758e4b 100644 --- a/sdks/java/extensions/sql/build.gradle +++ b/sdks/java/extensions/sql/build.gradle @@ -79,6 +79,7 @@ dependencies { provided project(":sdks:java:io:kafka") provided project(":sdks:java:io:google-cloud-platform") compile project(":sdks:java:io:mongodb") + compile library.java.avro provided project(":sdks:java:io:parquet") provided library.java.jackson_dataformat_xml provided library.java.hadoop_client diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java new file mode 100644 index 0000000..b2282ff --- /dev/null +++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java @@ -0,0 +1,132 @@ +/* + * 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.beam.sdk.extensions.sql.meta.provider.parquet; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.apache.avro.Schema; +import org.apache.avro.Schema.Field; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.annotations.Internal; +import org.apache.beam.sdk.extensions.sql.impl.BeamTableStatistics; +import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTableFilter; +import org.apache.beam.sdk.extensions.sql.meta.ProjectSupport; +import org.apache.beam.sdk.extensions.sql.meta.SchemaBaseBeamTable; +import org.apache.beam.sdk.extensions.sql.meta.Table; +import org.apache.beam.sdk.io.FileIO; +import org.apache.beam.sdk.io.parquet.ParquetIO; +import org.apache.beam.sdk.io.parquet.ParquetIO.Read; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.schemas.transforms.Convert; +import org.apache.beam.sdk.schemas.utils.AvroUtils; +import org.apache.beam.sdk.values.PBegin; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.Row; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Internal +@SuppressWarnings({"nullness"}) +class ParquetTable extends SchemaBaseBeamTable implements Serializable { + private static final Logger LOG = LoggerFactory.getLogger(ParquetTable.class); + + private final Table table; + + ParquetTable(Table table) { + super(table.getSchema()); + this.table = table; + } + + @Override + public PCollection buildIOReader(PBegin begin) { + final Schema schema = AvroUtils.toAvroSchema(table.getSchema()); + Read read = ParquetIO.read(schema).withBeamSchemas(true).from(table.getLocation() + "/*"); + return begin.apply("ParquetIORead", read).apply("ToRows", Convert.toRows()); + } + + @Override + public PCollection buildIOReader( + PBegin begin, BeamSqlTableFilter filters, List fieldNames) { + final Schema schema = AvroUtils.toAvroSchema(table.getSchema()); + Read read = ParquetIO.read(schema).withBeamSchemas(true).from(table.getLocation() + "/*"); + if (!fieldNames.isEmpty()) { + Schema projectionSchema = projectSchema(schema, fieldNames); + LOG.info("Projecting fields schema : " + projectionSchema.toString()); + read = read.withProjection(projectionSchema, projectionSchema); + } + return begin.apply("ParquetIORead", read).apply("ToRows", Convert.toRows()); + } + + /** Returns a copy of the {@link Schema} with only the fieldNames fields. */ + private static Schema projectSchema(Schema schema, List fieldNames) { + List selectedFields = new ArrayList<>(); + for (String fieldName : fieldNames) { + selectedFields.add(deepCopyField(schema.getField(fieldName))); + } + return Schema.createRecord( + schema.getName() + "_projected", + schema.getDoc(), + schema.getNamespace(), + schema.isError(), + selectedFields); + } + + private static Field deepCopyField(Field field) { + Schema.Field newField = + new Schema.Field( + field.name(), field.schema(), field.doc(), field.defaultVal(), field.order()); + for (Map.Entry kv : field.getObjectProps().entrySet()) { + newField.addProp(kv.getKey(), kv.getValue()); + } + if (field.aliases() != null) { + for (String alias : field.aliases()) { + newField.addAlias(alias); + } + } + return newField; + } + + @Override + public POutput buildIOWriter(PCollection input) { + final org.apache.avro.Schema schema = AvroUtils.toAvroSchema(input.getSchema()); + return input + .apply("ToGenericRecords", Convert.to(GenericRecord.class)) + .apply( + "ParquetIOWrite", + FileIO.write().via(ParquetIO.sink(schema)).to(table.getLocation())); + } + + @Override + public IsBounded isBounded() { + return PCollection.IsBounded.BOUNDED; + } + + @Override + public BeamTableStatistics getTableStatistics(PipelineOptions options) { + return BeamTableStatistics.BOUNDED_UNKNOWN; + } + + @Override + public ProjectSupport supportsProjects() { + return ProjectSupport.WITH_FIELD_REORDERING; + } +} diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProvider.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProvider.java index b8a55f5..f24e226 100644 --- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProvider.java +++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProvider.java @@ -18,18 +18,15 @@ package org.apache.beam.sdk.extensions.sql.meta.provider.parquet; import com.google.auto.service.AutoService; -import org.apache.beam.sdk.extensions.sql.meta.provider.SchemaIOTableProviderWrapper; +import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable; +import org.apache.beam.sdk.extensions.sql.meta.Table; +import org.apache.beam.sdk.extensions.sql.meta.provider.InMemoryMetaTableProvider; import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider; import org.apache.beam.sdk.io.parquet.ParquetIO; -import org.apache.beam.sdk.io.parquet.ParquetSchemaIOProvider; -import org.apache.beam.sdk.schemas.io.SchemaIOProvider; /** * {@link TableProvider} for {@link ParquetIO} for consumption by Beam SQL. * - *

Passes the {@link ParquetSchemaIOProvider} to the generalized table provider wrapper, {@link - * SchemaIOTableProviderWrapper}, for Parquet specific behavior. - * *

A sample of parquet table is: * * 

{@code
@@ -39,19 +36,18 @@ import org.apache.beam.sdk.schemas.io.SchemaIOProvider;
  *   favorite_numbers ARRAY
  * )
  * TYPE 'parquet'
- * LOCATION '/home/admin/users.parquet'
+ * LOCATION '/home/admin/orders/'
  * }
*/ @AutoService(TableProvider.class) -public class ParquetTableProvider extends SchemaIOTableProviderWrapper { +public class ParquetTableProvider extends InMemoryMetaTableProvider { @Override - public SchemaIOProvider getSchemaIOProvider() { - return new ParquetSchemaIOProvider(); + public String getTableType() { + return "parquet"; } - // TODO[BEAM-10516]: remove this override after TableProvider problem is fixed @Override - public String getTableType() { - return "parquet"; + public BeamSqlTable buildBeamSqlTable(Table table) { + return new ParquetTable(table); } } diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderTest.java index 49d7b6e..71680f7 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderTest.java +++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTableProviderTest.java @@ -44,7 +44,13 @@ public class ParquetTableProviderTest { private static final String FIELD_NAMES = "(name VARCHAR, age BIGINT, country VARCHAR)"; - private static final Schema OUTPUT_ROW_SCHEMA = + private static final Schema TABLE_SCHEMA = + Schema.builder() + .addStringField("name") + .addInt64Field("age") + .addStringField("country") + .build(); + private static final Schema PROJECTED_SCHEMA = Schema.builder().addInt64Field("age").addStringField("country").build(); @Test @@ -61,15 +67,34 @@ public class ParquetTableProviderTest { writePipeline, env.parseQuery( "INSERT INTO PersonInfo VALUES ('Alan', 22, 'England'), ('John', 42, 'USA')")); - writePipeline.run().waitUntilFinish(); PCollection rows = + BeamSqlRelUtils.toPCollection(readPipeline, env.parseQuery("SELECT * FROM PersonInfo")); + PAssert.that(rows) + .containsInAnyOrder( + Row.withSchema(TABLE_SCHEMA).addValues("Alan", 22L, "England").build(), + Row.withSchema(TABLE_SCHEMA).addValues("John", 42L, "USA").build()); + + PCollection filtered = BeamSqlRelUtils.toPCollection( - readPipeline, env.parseQuery("SELECT age, country FROM PersonInfo WHERE age > 25")); + readPipeline, env.parseQuery("SELECT * FROM PersonInfo WHERE age > 25")); + PAssert.that(filtered) + .containsInAnyOrder(Row.withSchema(TABLE_SCHEMA).addValues("John", 42L, "USA").build()); - PAssert.that(rows) - .containsInAnyOrder(Row.withSchema(OUTPUT_ROW_SCHEMA).addValues(42L, "USA").build()); + PCollection projected = + BeamSqlRelUtils.toPCollection( + readPipeline, env.parseQuery("SELECT age, country FROM PersonInfo")); + PAssert.that(projected) + .containsInAnyOrder( + Row.withSchema(PROJECTED_SCHEMA).addValues(22L, "England").build(), + Row.withSchema(PROJECTED_SCHEMA).addValues(42L, "USA").build()); + + PCollection filteredAndProjected = + BeamSqlRelUtils.toPCollection( + readPipeline, env.parseQuery("SELECT age, country FROM PersonInfo WHERE age > 25")); + PAssert.that(filteredAndProjected) + .containsInAnyOrder(Row.withSchema(PROJECTED_SCHEMA).addValues(42L, "USA").build()); PipelineResult.State state = readPipeline.run().waitUntilFinish(); assertEquals(State.DONE, state); diff --git a/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetSchemaIOProvider.java b/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetSchemaIOProvider.java deleted file mode 100644 index 71ef5e2..0000000 --- a/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetSchemaIOProvider.java +++ /dev/null @@ -1,127 +0,0 @@ -/* - * 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.beam.sdk.io.parquet; - -import com.google.auto.service.AutoService; -import java.io.Serializable; -import org.apache.avro.generic.GenericRecord; -import org.apache.beam.sdk.annotations.Internal; -import org.apache.beam.sdk.io.FileIO; -import org.apache.beam.sdk.schemas.Schema; -import org.apache.beam.sdk.schemas.io.SchemaIO; -import org.apache.beam.sdk.schemas.io.SchemaIOProvider; -import org.apache.beam.sdk.schemas.transforms.Convert; -import org.apache.beam.sdk.schemas.utils.AvroUtils; -import org.apache.beam.sdk.transforms.PTransform; -import org.apache.beam.sdk.values.PBegin; -import org.apache.beam.sdk.values.PCollection; -import org.apache.beam.sdk.values.PDone; -import org.apache.beam.sdk.values.POutput; -import org.apache.beam.sdk.values.Row; - -/** - * An implementation of {@link SchemaIOProvider} for reading and writing parquet files with {@link - * ParquetIO}. - */ -@Internal -@AutoService(SchemaIOProvider.class) -@SuppressWarnings({ - "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402) -}) -public class ParquetSchemaIOProvider implements SchemaIOProvider { - /** Returns an id that uniquely represents this IO. */ - @Override - public String identifier() { - return "parquet"; - } - - /** - * Returns the expected schema of the configuration object. Note this is distinct from the schema - * of the data source itself. No configuration expected for parquet. - */ - @Override - public Schema configurationSchema() { - return Schema.builder().build(); - } - - /** - * Produce a SchemaIO given a String representing the data's location, the schema of the data that - * resides there, and some IO-specific configuration object. - */ - @Override - public ParquetSchemaIO from(String location, Row configuration, Schema dataSchema) { - return new ParquetSchemaIO(location, dataSchema); - } - - @Override - public boolean requiresDataSchema() { - return true; - } - - @Override - public PCollection.IsBounded isBounded() { - return PCollection.IsBounded.BOUNDED; - } - - /** An abstraction to create schema aware IOs. */ - private static class ParquetSchemaIO implements SchemaIO, Serializable { - protected final Schema dataSchema; - protected final String location; - - private ParquetSchemaIO(String location, Schema dataSchema) { - this.dataSchema = dataSchema; - this.location = location; - } - - @Override - public Schema schema() { - return dataSchema; - } - - @Override - public PTransform> buildReader() { - return new PTransform>() { - @Override - public PCollection expand(PBegin begin) { - org.apache.avro.Schema schema = AvroUtils.toAvroSchema(dataSchema); - return begin - .apply( - "ParquetIORead", - ParquetIO.read(schema).withBeamSchemas(true).from(location + "/*")) - .apply("ToRows", Convert.toRows()); - } - }; - } - - @Override - public PTransform, POutput> buildWriter() { - return new PTransform, POutput>() { - @Override - public PDone expand(PCollection input) { - final org.apache.avro.Schema schema = AvroUtils.toAvroSchema(input.getSchema()); - input - .apply("ToGenericRecords", Convert.to(GenericRecord.class)) - .apply( - "ParquetIOWrite", - FileIO.write().via(ParquetIO.sink(schema)).to(location)); - return PDone.in(input.getPipeline()); - } - }; - } - } -}