ahmedabu98 commented on code in PR #40143:
URL: https://github.com/apache/beam/pull/40143#discussion_r4030506947
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java:
##########
@@ -949,4 +963,609 @@ private DataWriter<Record> createWriter(String file,
@Nullable StructLike partit
private Record record(int id, String name, int age) {
return GenericRecord.create(icebergSchema).copy("id", id, "name", name,
"age", age);
}
+
+ // ---- ConvertToDataFile coverage check and pinned columns
+
+ private static final SchemaEvolutionConfig ADDITIONS =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
+
+ private PCollectionTuple convert(SchemaEvolutionConfig config, String...
files) {
+ PCollectionTuple out =
+ pipeline
+ .apply("Create Input", Create.of(Arrays.asList(files)))
+ .apply(
+ ParDo.of(
+ new AddFiles.ConvertToDataFile(
+ catalogConfig, tableId.toString(), null, null,
null, null, config))
+ .withOutputTags(
+ AddFiles.ConvertToDataFile.DATA_FILES,
+ TupleTagList.of(AddFiles.ConvertToDataFile.ERRORS)));
+
out.get(AddFiles.ConvertToDataFile.ERRORS).setRowSchema(AddFiles.ERROR_SCHEMA);
+ return out;
+ }
+
+ private String writeWithSchema(String name, Schema schema, Record...
records) throws IOException {
+ String file = root + name;
+ DataWriter<Record> writer =
+ Parquet.writeData(Files.localOutput(file))
+ .schema(schema)
+ .withSpec(PartitionSpec.unpartitioned())
+ .createWriterFunc(GenericParquetWriter::create)
+ .build();
+ try {
+ for (Record record : records) {
+ writer.write(record);
+ }
+ } finally {
+ writer.close();
+ }
+ return file;
+ }
+
+ private String writeOneRecord(String name) throws IOException {
+ String file = root + name;
+ DataWriter<Record> writer = createWriter(file);
+ writer.write(record(1, "a", 1));
+ writer.close();
+ return file;
+ }
+
+ private static final Schema WIDER =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.StringType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()),
+ Types.NestedField.optional(4, "email", Types.StringType.get()));
+
+ private String writeWider(String name) throws IOException {
+ Record record = GenericRecord.create(WIDER);
+ record.setField("id", 1);
+ record.setField("name", "a");
+ record.setField("age", 1);
+ record.setField("email", "e");
+ return writeWithSchema(name, WIDER, record);
+ }
+
+ private void assertSingleError(PCollectionTuple out, String file, String
contains) {
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.DATA_FILES)).empty();
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ Row row = Iterables.getOnlyElement(rows);
+ assertEquals(file, row.getString("file"));
+ assertThat(row.getString("error"), containsString(contains));
+ return null;
+ });
+ }
+
+ private void assertRegisters(PCollectionTuple out, long files) {
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS)).empty();
+
PAssert.thatSingleton(out.get(AddFiles.ConvertToDataFile.DATA_FILES).apply(Count.globally()))
+ .isEqualTo(files);
+ }
+
+ @Test
+ public void testEmptySchemaTableWarnsAndStillRegisters() throws Exception {
+ catalog.createTable(tableId, new Schema());
+ String file = writeOneRecord("data.parquet");
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ logs.verifyWarn("has no columns");
+ }
+
+ /** Files without embedded field ids against a zero-column table still
register. */
+ @Test
+ public void testEmptySchemaTableWithIdLessParquetStillRegisters() throws
Exception {
+ catalog.createTable(tableId, new Schema());
+ File file = new File(temp.getRoot(), "idless.parquet");
+ org.apache.avro.Schema avro =
+ org.apache.avro.SchemaBuilder.record("r")
+ .fields()
+ .requiredInt("id")
+ .optionalString("name")
+ .name("address")
+ .type()
+ .record("address")
+ .fields()
+ .optionalString("city")
+ .endRecord()
+ .noDefault()
+ .endRecord();
+ try (org.apache.parquet.hadoop.ParquetWriter<Object> writer =
+ org.apache.parquet.avro.AvroParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(file.getAbsolutePath()))
+ .withSchema(avro)
+ .build()) {
+ org.apache.avro.generic.GenericData.Record record =
+ new org.apache.avro.generic.GenericData.Record(avro);
+ record.put("id", 1);
+ record.put("name", "a");
+ org.apache.avro.generic.GenericData.Record address =
+ new
org.apache.avro.generic.GenericData.Record(avro.getField("address").schema());
+ address.put("city", "c");
+ record.put("address", address);
+ writer.write(record);
+ }
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(),
file.getAbsolutePath());
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testCoveredFileRegistersWithEvolutionEnabled() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeOneRecord("data.parquet");
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testUncoveredFileRoutesToErrorsWhenEvolutionEnabled() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeWider("wider.parquet");
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertSingleError(out, file, "does not cover the file");
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ assertThat(
+ Iterables.getOnlyElement(rows).getString("error"),
+ containsString("add optional email string"));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testExtraColumnsRegisterWhenEvolutionDisabled() throws Exception
{
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeWider("wider.parquet");
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testUnreadableSchemaRoutesToErrorsWithConverterMessage() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ // a legacy unannotated repeated field: readable Parquet, rejected by
Iceberg's converter
+ org.apache.parquet.schema.MessageType legacy =
+ org.apache.parquet.schema.Types.buildMessage()
+
.required(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32)
+ .named("id")
+
.repeated(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32)
+ .named("vals")
+ .named("root");
+ File legacyFile = new File(temp.getRoot(), "legacy.parquet");
+ try
(org.apache.parquet.hadoop.ParquetWriter<org.apache.parquet.example.data.Group>
writer =
+ org.apache.parquet.hadoop.example.ExampleParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(legacyFile.getAbsolutePath()))
+ .withType(legacy)
+ .build()) {
+ org.apache.parquet.example.data.Group group =
+ new
org.apache.parquet.example.data.simple.SimpleGroupFactory(legacy).newGroup();
+ group.add("id", 1);
+ group.add("vals", 2);
+ writer.write(group);
+ }
+ String file = legacyFile.getAbsolutePath();
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertSingleError(out, file,
AddFiles.ConvertToDataFile.UNREADABLE_SCHEMA_ERROR);
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ assertThat(
+ Iterables.getOnlyElement(rows).getString("error"),
+ containsString("repetition REPEATED"));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ }
+
+ // ---- pinned columns
+
+ private static SchemaEvolutionConfig pinned(String column) {
+ return SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton(column))
+ .build();
+ }
+
+ private static final Schema OPTIONAL_NAME =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()));
+
+ private static final Schema WITHOUT_NAME =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()));
+
+ private String writeCleanName(String name) throws IOException {
+ return writeWithSchema(
+ name,
+ OPTIONAL_NAME,
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 1, "name", "a", "age",
1));
+ }
+
+ private String writeOneNullName(String name) throws IOException {
+ return writeWithSchema(
+ name,
+ OPTIONAL_NAME,
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 1, "name", "a", "age",
1),
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 2, "age", 2));
+ }
+
+ private String writeWithoutName(String name) throws IOException {
+ return writeWithSchema(
+ name, WITHOUT_NAME, GenericRecord.create(WITHOUT_NAME).copy("id", 1,
"age", 1));
+ }
+
+ @Test
+ public void testPinnedColumnWithNullsRoutesToErrors() throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
Review Comment:
These Pin tests should use required columns right? same with the tests below
that use `WITH_ITEMS` and `WITH_ADDRESS`, they currently all pin optional
columns
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -464,19 +560,139 @@ private Callable<ProcessResult> createProcessTask(
.withPartitionPath(partitionPath)
.build();
return new ProcessResult(
- SerializableDataFile.from(df, table.spec()), null, timestamp,
window, paneInfo);
+ SerializableDataFile.from(df, table.spec()),
+ null,
+ verdict.unverified,
+ timestamp,
+ window,
+ paneInfo);
} catch (Exception e) {
// getLength is a per-file read (e.g. the file was deleted
mid-flight).
return errorResult(filePath, errorMessage(e), timestamp, window,
paneInfo);
}
};
}
+ /**
+ * The checks the options promise, in order: a format the checks can read,
a convertible schema,
+ * coverage by the table, pins. The first failure is the verdict.
+ */
+ private Verdict verify(String filePath, FileFormat format, @Nullable
ParquetMetadata footer) {
+ if (!format.equals(FileFormat.PARQUET)) {
+ if (evolution.getUnverifiableFileHandling() ==
UnverifiableFileHandling.REJECT) {
+ return Verdict.error(UNCHECKED_FORMAT_ERROR + format.name());
+ }
+ if (firstTime("unchecked format")) {
+ LOG.warn(
+ "Registering {} files in table {} unchecked
(UnverifiableFileHandling.ACCEPT):"
+ + " coverage and pin checks read only Parquet, so a required
column such a file"
+ + " lacks or holds nulls in fails reads of the table, not
registration. First"
+ + " file: {}",
+ format,
+ identifier,
+ filePath);
+ }
+ return Verdict.unverified(Unverified.FORMAT);
+ }
+ ParquetMetadata parquetFooter = checkStateNotNull(footer, "Parquet
checks need the footer");
+ org.apache.iceberg.Schema fileSchema;
+ try {
+ fileSchema = FileSchemas.effective(parquetFooter);
+ } catch (Exception e) {
+ return Verdict.error(UNREADABLE_SCHEMA_ERROR + errorMessage(e));
+ }
+ @Nullable String uncovered = uncoveredReason(fileSchema);
+ if (uncovered != null) {
+ return Verdict.error(uncovered);
+ }
+ return checkPins(filePath, fileSchema, parquetFooter);
+ }
+
+ /**
+ * The pre-pass commits the schema before paths reach this stage, so the
cached table normally
+ * covers every file. If not, refresh once (a commit may have landed since
the table was cached)
+ * and report the remaining delta. Never changes the schema.
+ */
+ private @Nullable String uncoveredReason(org.apache.iceberg.Schema
fileSchema) {
+ Table table = checkStateNotNull(this.table);
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ return null;
+ }
+ synchronized (this) {
+ table.refresh();
+ }
+ delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ return null;
+ }
+ String reason = delta.disallowedReason(evolution);
+ if (reason.isEmpty()) {
+ reason = "changes not applied: " + String.join("; ",
delta.descriptions());
+ }
+ return UNCOVERED_ERROR + reason;
+ }
+
+ /**
+ * A pinned column must be present and provably null-free; a zero-row file
is vacuously fine.
+ * The evidence is the footer's own null counts read by the tighten rules
({@link
+ * FileSchemas#nullCount}), never the Metrics built for the DataFile: the
table's
+ * write.metadata.metrics configuration shapes those (mode none, or the
inferred-column cap on
+ * wide schemas, drops the counts) and must not be able to turn pin
enforcement off. A pin with
+ * no count is a violation under REJECT; under ACCEPT it is recorded as
unproven and the walk
+ * goes on, so a pin the footer does count nulls for still fails the file.
+ */
+ private Verdict checkPins(
+ String filePath, org.apache.iceberg.Schema fileSchema, ParquetMetadata
footer) {
+ Table table = checkStateNotNull(this.table);
+ List<String> unproven = new ArrayList<>();
+ for (String pinned : evolution.getRequiredColumns()) {
+ if (table.schema().findField(pinned) == null) {
+ continue;
+ }
+ if (fileSchema.findField(pinned) == null) {
+ return Verdict.error(PINNED_COLUMN_ERROR + pinned + " is absent from
the file");
+ }
+ @Nullable Long nulls = FileSchemas.nullCount(footer, fileSchema,
pinned);
+ if (nulls == null) {
+ if (evolution.getUnverifiableFileHandling() ==
UnverifiableFileHandling.REJECT) {
+ return Verdict.error(
+ PINNED_COLUMN_ERROR + pinned + " has no null count statistics
in the file");
+ }
+ unproven.add(pinned);
+ continue;
Review Comment:
We should also trust when the Parquet file tells us the column is required
right? AFAIK a required column doesn't have to provide null counts
--
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]
