jonathansudhakar1 opened a new pull request, #40970: URL: https://github.com/apache/beam/pull/40970
Fixes #20758. The [CREATE EXTERNAL TABLE](https://beam.apache.org/documentation/dsls/sql/extensions/create-external-table/) page has documented `TYPE parquet` for a while, but not `TYPE avro`, even though Beam SQL registers an Avro table provider (`AvroTableProvider`, backed by `AvroSchemaIOProvider`). This PR: - adds `avro` to the page's list of table types; - adds an **Avro** section after Parquet, in the same layout: Syntax, Read Mode, Write Mode, Schema and Example. The section describes `AvroSchemaIOProvider` at `ae065d1302a`: - **Read:** `AvroIO.readGenericRecords(<table schema>).from(LOCATION)`, so `LOCATION` can be a file or a file pattern. - **Write, bounded input:** `AvroIO.writeGenericRecords(...).to(LOCATION).withoutSharding()`, which writes a single file at `LOCATION`. - **Write, windowed:** for unbounded input, or when the `writeWindowSizeSeconds` table property is set, it uses fixed windows with `withWindowedWrites().withNumShards(1)`. That writes one file per window, with `LOCATION` as the prefix. The default window for unbounded input is one minute. ## Testing Docs-only change, so no CHANGES.md entry. There's no local Hugo, so I haven't rendered the page. The new section uses only plain Markdown (headings, lists and code fences), like the surrounding sections. To check that every statement in the new section is accurate, I wrote a temporary JUnit test (not part of this PR; source below). It runs the documented DDL through `SqlTransform` on the DirectRunner. It covers: 1. **The example as documented:** `INSERT INTO` a bounded table writes exactly one file named by `LOCATION` (`[person-info.avro]`), and `SELECT` reads it back. 2. **Glob reads:** rows written to `a.avro` and `b.avro` are both read back through a table with `LOCATION '<dir>/*.avro'`. 3. **`TBLPROPERTIES '{"writeWindowSizeSeconds": 60}'` on bounded input:** gives windowed output, a single file named `people-<window>-pane-0-last-00000-of-00001`. Its rows are read back through `LOCATION '<prefix>*'`. 4. **Files written by other tools:** a file written directly with Avro's `DataFileWriter` reads correctly into the table. Its schema has the fields in a different order and non-nullable types, so fields are resolved by name. ``` source .contrib/env.sh # JDK 21 ./gradlew :sdks:java:extensions:sql:test \ --tests 'org.apache.beam.sdk.extensions.sql.meta.provider.avro.*' -PenableCheckerFramework=false ``` `RESULT [:sdks:java:extensions:sql:test]: SUCCESS - 5 tests, 5 passed, 0 failed, 0 skipped` The 5 tests are the existing `AvroTableProviderTest` plus the 4 checks above. Checker Framework was off because no main sources change. The only claim the test doesn't exercise is the one-minute default window for unbounded input. It comes straight from `AvroSchemaIOProvider.buildWriter()` (`Duration.standardMinutes(1)` when `writeWindowSizeSeconds` is unset). <details> <summary>Temporary verification test (not committed)</summary> ```java /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package org.apache.beam.sdk.extensions.sql.meta.provider.avro; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.io.File; import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; import org.apache.beam.sdk.extensions.sql.SqlTransform; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.Row; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; /** Temporary checks of the behavior described in the CREATE EXTERNAL TABLE docs. Not committed. */ @RunWith(JUnit4.class) public class ZzAvroDocsVerificationTest { @Rule public TestPipeline p1 = TestPipeline.create(); @Rule public TestPipeline p2 = TestPipeline.create(); @Rule public TestPipeline p3 = TestPipeline.create(); @Rule public TemporaryFolder tmp = new TemporaryFolder(); private static final Schema OUT = Schema.builder() .addStringField("name") .addInt64Field("age") .addStringField("country") .build(); private static String ddl(String table, String location, String extra) { return "CREATE EXTERNAL TABLE " + table + " (\n name VARCHAR,\n age BIGINT,\n country VARCHAR\n)\nTYPE avro\nLOCATION '" + location + "'" + extra; } private static List<String> files(File dir) { return Arrays.stream(dir.listFiles()).map(File::getName).sorted().collect(Collectors.toList()); } /** The documented example: a bounded write produces exactly one file at LOCATION. */ @Test public void docExampleWritesOneFileAtLocationAndReadsItBack() { File loc = new File(tmp.getRoot(), "person-info.avro"); String ddl = ddl("person_info", loc.getAbsolutePath(), ""); p1.apply( SqlTransform.query( "INSERT INTO person_info VALUES ('Alan', 22, 'England'), ('John', 42, 'USA')") .withDdlString(ddl)); p1.run().waitUntilFinish(); System.out.println("DOCCHECK single-file write produced: " + files(tmp.getRoot())); assertEquals(Arrays.asList("person-info.avro"), files(tmp.getRoot())); PCollection<Row> rows = p2.apply( SqlTransform.query("SELECT name, age, country FROM person_info WHERE age > 25") .withDdlString(ddl)); PAssert.that(rows) .containsInAnyOrder(Row.withSchema(OUT).addValues("John", 42L, "USA").build()); p2.run().waitUntilFinish(); } /** LOCATION can be a glob for reads. */ @Test public void readsAllFilesMatchingGlob() { File dir = tmp.getRoot(); p1.apply( SqlTransform.query("INSERT INTO a VALUES ('Alan', 22, 'England')") .withDdlString(ddl("a", new File(dir, "a.avro").getAbsolutePath(), ""))); p1.run().waitUntilFinish(); p2.apply( SqlTransform.query("INSERT INTO b VALUES ('John', 42, 'USA')") .withDdlString(ddl("b", new File(dir, "b.avro").getAbsolutePath(), ""))); p2.run().waitUntilFinish(); PCollection<Row> rows = p3.apply( SqlTransform.query("SELECT name, age, country FROM all_people") .withDdlString(ddl("all_people", dir.getAbsolutePath() + "/*.avro", ""))); PAssert.that(rows) .containsInAnyOrder( Row.withSchema(OUT).addValues("Alan", 22L, "England").build(), Row.withSchema(OUT).addValues("John", 42L, "USA").build()); p3.run().waitUntilFinish(); } /** writeWindowSizeSeconds: windowed writes use LOCATION as a filename prefix. */ @Test public void windowedWriteUsesLocationAsPrefix() { File dir = tmp.getRoot(); String prefix = new File(dir, "people").getAbsolutePath(); p1.apply( SqlTransform.query("INSERT INTO people VALUES ('Alan', 22, 'England'), ('John', 42, 'USA')") .withDdlString( ddl("people", prefix, "\nTBLPROPERTIES '{\"writeWindowSizeSeconds\": 60}'"))); p1.run().waitUntilFinish(); List<String> written = files(dir); System.out.println("DOCCHECK windowed write produced: " + written); List<String> outputs = written.stream().filter(n -> !n.startsWith(".")).collect(Collectors.toList()); assertEquals(1, outputs.size()); assertTrue(outputs.get(0).startsWith("people-") && outputs.get(0).endsWith("-of-00001")); PCollection<Row> rows = p2.apply( SqlTransform.query("SELECT name, age, country FROM people_read") .withDdlString(ddl("people_read", prefix + "*", ""))); PAssert.that(rows) .containsInAnyOrder( Row.withSchema(OUT).addValues("Alan", 22L, "England").build(), Row.withSchema(OUT).addValues("John", 42L, "USA").build()); p2.run().waitUntilFinish(); } /** Files written by other tools: Avro schema resolution matches fields by name. */ @Test public void readsFileWithDifferentFieldOrderAndNonNullTypes() throws Exception { org.apache.avro.Schema writerSchema = new org.apache.avro.Schema.Parser() .parse( "{\"type\":\"record\",\"name\":\"Person\",\"fields\":[" + "{\"name\":\"country\",\"type\":\"string\"}," + "{\"name\":\"age\",\"type\":\"long\"}," + "{\"name\":\"name\",\"type\":\"string\"}]}"); File f = new File(tmp.getRoot(), "external.avro"); try (org.apache.avro.file.DataFileWriter<org.apache.avro.generic.GenericRecord> w = new org.apache.avro.file.DataFileWriter<>( new org.apache.avro.generic.GenericDatumWriter<org.apache.avro.generic.GenericRecord>( writerSchema))) { w.create(writerSchema, f); org.apache.avro.generic.GenericRecord r = new org.apache.avro.generic.GenericData.Record(writerSchema); r.put("country", "USA"); r.put("age", 42L); r.put("name", "John"); w.append(r); } PCollection<Row> rows = p1.apply( SqlTransform.query("SELECT name, age, country FROM ext") .withDdlString(ddl("ext", f.getAbsolutePath(), ""))); PAssert.that(rows) .containsInAnyOrder(Row.withSchema(OUT).addValues("John", 42L, "USA").build()); p1.run().waitUntilFinish(); } } ``` </details> 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
