ahmedabu98 commented on code in PR #39975:
URL: https://github.com/apache/beam/pull/39975#discussion_r3925885954
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java:
##########
@@ -18,29 +18,208 @@
package org.apache.beam.sdk.io.iceberg;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
/**
- * Derives the schema a file contributes to schema inference. The canonical
form sorts struct fields
- * by name at every level and renumbers ids in deterministic order, so files
that differ only in
- * column order produce identical JSON. Ids are positional and meaningless:
the commit side
- * reconciles columns by name.
+ * What a file contributes to schema inference: the canonical form of the
schema it declares, and
+ * the columns its footer proves free of nulls.
+ *
+ * <p>The schema half depends only on the declared schema, never on the data,
so files written by
+ * the same job dedup to one entry no matter where their nulls fall. The null
evidence is combined
+ * per schema by {@link CollectDistinctSchemas} and reapplied by the commit
side via {@link
+ * #markRequired}.
+ *
+ * <p>The canonical form sorts struct fields by name at every level and
renumbers ids. Ids are
+ * positional and meaningless (the commit side reconciles columns by name), so
never diff two file
+ * schemas by id. Other field attributes (doc, defaults) are preserved,
matching what SchemaDelta
+ * compares. Column paths are dotted, like pins.
*/
final class FileSchemas {
private FileSchemas() {}
+ /** Canonical JSON of the schema the file declares. */
static String canonicalJson(ParquetMetadata footer) {
Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
return SchemaParser.toJson(canonical(converted));
}
+ /** This file as a schema group of one: its declared schema and its
null-free columns. */
+ static CollectDistinctSchemas.SchemaGroup schemaGroup(ParquetMetadata
footer) {
+ Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+ Schema tightened = tighten(converted, footer);
+ return new CollectDistinctSchemas.SchemaGroup(
+ SchemaParser.toJson(canonical(converted)), 1,
changedToRequired(converted, tightened));
+ }
+
+ /**
+ * Marks a declared-optional column required when every row group has a null
count of zero for it,
+ * so the file does not request a relaxation it does not need; an absent
count is not proof. A
+ * struct is null-free when any leaf under it is (a null struct nulls all
its leaves). Nothing
+ * under lists or maps is tightened: a zero count there would be valid
evidence too, but mapping
+ * physical chunk paths (writer-dependent names like {@code list.element},
{@code array}) onto the
+ * converted schema is not worth it. A file with no row groups has no rows
and proves every
+ * column, matching how the pin check treats empty files.
+ */
+ static Schema tighten(Schema schema, ParquetMetadata footer) {
+ if (footer.getBlocks().isEmpty()) {
+ return new Schema(tightenAll(schema.asStruct()).fields());
+ }
+ Set<List<String>> zeroNullLeaves = leafPathsWithZeroNullCounts(footer);
+ if (zeroNullLeaves.isEmpty()) {
+ return schema;
+ }
+ return new Schema(
+ tightenStruct(schema.asStruct(), new ArrayList<>(),
zeroNullLeaves).struct.fields());
+ }
+
+ /** With no rows, nothing can hold a null: every leaf and struct outside
lists and maps. */
+ private static Types.StructType tightenAll(Types.StructType struct) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ Type type = field.type();
+ if (type.isStructType()) {
+ fields.add(withOptionality(field, tightenAll(type.asStructType()),
false));
+ } else if (type.isPrimitiveType()) {
+ fields.add(withOptionality(field, type, false));
+ } else {
+ fields.add(field);
+ }
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /**
+ * Returns the schema with the given dotted column paths made required. The
commit side parses a
+ * group's schema JSON (optionality as the writer declared it) and applies
the group's null-free
+ * columns with this before classifying, so only relaxations some file
actually needs remain.
+ */
+ static Schema markRequired(Schema declared, Collection<String> columns) {
+ if (columns.isEmpty()) {
+ return declared;
+ }
+ Types.StructType required = markRequiredStruct(declared.asStruct(), "",
new HashSet<>(columns));
+ return new Schema(required.fields());
+ }
+
+ private static Types.StructType markRequiredStruct(
+ Types.StructType struct, String prefix, Set<String> columns) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ String path = prefix + field.name();
+ Type type = field.type();
+ if (type.isStructType()) {
+ type = markRequiredStruct(type.asStructType(), path + ".", columns);
+ }
+ boolean required = !field.isRequired() && columns.contains(path);
+ fields.add(withOptionality(field, type, field.isOptional() &&
!required));
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /** Dotted paths of fields the tightened schema made required, sorted. */
+ private static List<String> changedToRequired(Schema declared, Schema
tightened) {
+ List<String> paths = new ArrayList<>();
+ collectChangedToRequired(declared.asStruct(), tightened.asStruct(), "",
paths);
+ Collections.sort(paths);
+ return paths;
+ }
+
+ private static void collectChangedToRequired(
+ Types.StructType declared, Types.StructType tightened, String prefix,
List<String> out) {
+ for (int i = 0; i < declared.fields().size(); i++) {
+ Types.NestedField before = declared.fields().get(i);
+ Types.NestedField after = tightened.fields().get(i);
+ String path = prefix + before.name();
+ if (before.isOptional() && after.isRequired()) {
+ out.add(path);
+ }
Review Comment:
Not sure how `nullFreeColumns` is used downstream, but should it also
include Required fields?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java:
##########
@@ -17,87 +17,251 @@
*/
package org.apache.beam.sdk.io.iceberg;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.TreeMap;
+import java.util.TreeSet;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderRegistry;
-import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.CustomCoder;
import org.apache.beam.sdk.coders.ListCoder;
import org.apache.beam.sdk.coders.MapCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.transforms.Combine;
-import org.apache.beam.sdk.values.KV;
+import org.checkerframework.checker.nullness.qual.Nullable;
/**
- * Collects the distinct schemas among canonical file schema JSONs (see {@link
FileSchemas}), with
- * the number of files per schema, most common first (ties broken by JSON).
The commit side applies
- * schemas in this order, so the schema covering the most files wins a
conflict.
- *
- * <p>Inputs are compared as strings, so they must already be canonical.
+ * One output entry per distinct schema: its file count and the columns EVERY
file carrying it
+ * proved free of nulls; one file with a null in "name" forces "name" to
relax, however many clean
+ * files sit next to it. Entries come out most common first (ties broken by
the JSON text) because
+ * the commit side applies schemas in that order and the most common schema
should win a conflict.
+ * Schemas are compared as strings, so inputs must already be canonical.
*/
class CollectDistinctSchemas
- extends Combine.CombineFn<String, Map<String, Long>, List<KV<String,
Long>>> {
+ extends Combine.CombineFn<
+ CollectDistinctSchemas.SchemaGroup,
+ Map<String, CollectDistinctSchemas.Group>,
+ List<CollectDistinctSchemas.SchemaGroup>> {
+
+ /** Mutable accumulator counterpart of {@link SchemaGroup}. */
+ static final class Group {
+ long files;
+ TreeSet<String> nullFreeColumns;
+
+ Group(long files, TreeSet<String> nullFreeColumns) {
+ this.files = files;
+ this.nullFreeColumns = nullFreeColumns;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object other) {
+ if (!(other instanceof Group)) {
+ return false;
+ }
+ Group that = (Group) other;
+ return files == that.files &&
nullFreeColumns.equals(that.nullFreeColumns);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(files, nullFreeColumns);
+ }
+ }
+
+ /**
+ * A schema, how many files carry it, and the columns all of them proved
free of nulls.
+ * ReadFooterSchema emits one per file ({@code files} = 1); this combiner
merges them.
+ */
+ static final class SchemaGroup {
Review Comment:
Could bypass a lot of boiler plate (equals, hashCode, custom coder) by using
`@AutoValue` here
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java:
##########
@@ -17,87 +17,251 @@
*/
package org.apache.beam.sdk.io.iceberg;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.TreeMap;
+import java.util.TreeSet;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderRegistry;
-import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.CustomCoder;
import org.apache.beam.sdk.coders.ListCoder;
import org.apache.beam.sdk.coders.MapCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.transforms.Combine;
-import org.apache.beam.sdk.values.KV;
+import org.checkerframework.checker.nullness.qual.Nullable;
/**
- * Collects the distinct schemas among canonical file schema JSONs (see {@link
FileSchemas}), with
- * the number of files per schema, most common first (ties broken by JSON).
The commit side applies
- * schemas in this order, so the schema covering the most files wins a
conflict.
- *
- * <p>Inputs are compared as strings, so they must already be canonical.
+ * One output entry per distinct schema: its file count and the columns EVERY
file carrying it
+ * proved free of nulls; one file with a null in "name" forces "name" to
relax, however many clean
+ * files sit next to it. Entries come out most common first (ties broken by
the JSON text) because
+ * the commit side applies schemas in that order and the most common schema
should win a conflict.
+ * Schemas are compared as strings, so inputs must already be canonical.
*/
class CollectDistinctSchemas
- extends Combine.CombineFn<String, Map<String, Long>, List<KV<String,
Long>>> {
+ extends Combine.CombineFn<
+ CollectDistinctSchemas.SchemaGroup,
+ Map<String, CollectDistinctSchemas.Group>,
+ List<CollectDistinctSchemas.SchemaGroup>> {
+
+ /** Mutable accumulator counterpart of {@link SchemaGroup}. */
+ static final class Group {
+ long files;
+ TreeSet<String> nullFreeColumns;
+
+ Group(long files, TreeSet<String> nullFreeColumns) {
+ this.files = files;
+ this.nullFreeColumns = nullFreeColumns;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object other) {
+ if (!(other instanceof Group)) {
+ return false;
+ }
+ Group that = (Group) other;
+ return files == that.files &&
nullFreeColumns.equals(that.nullFreeColumns);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(files, nullFreeColumns);
+ }
+ }
+
+ /**
+ * A schema, how many files carry it, and the columns all of them proved
free of nulls.
+ * ReadFooterSchema emits one per file ({@code files} = 1); this combiner
merges them.
+ */
+ static final class SchemaGroup {
+ final String schemaJson;
+ final long files;
+ final List<String> nullFreeColumns;
+
+ SchemaGroup(String schemaJson, long files, List<String> nullFreeColumns) {
+ this.schemaJson = schemaJson;
+ this.files = files;
+ this.nullFreeColumns = nullFreeColumns;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object other) {
+ if (!(other instanceof SchemaGroup)) {
+ return false;
+ }
+ SchemaGroup that = (SchemaGroup) other;
+ return files == that.files
+ && schemaJson.equals(that.schemaJson)
+ && nullFreeColumns.equals(that.nullFreeColumns);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(schemaJson, files, nullFreeColumns);
+ }
+
+ @Override
+ public String toString() {
+ return files + " file(s), null-free in " + nullFreeColumns + ", schema "
+ schemaJson;
+ }
+ }
@Override
- public Map<String, Long> createAccumulator() {
+ public Map<String, Group> createAccumulator() {
return new TreeMap<>();
}
@Override
- public Map<String, Long> addInput(Map<String, Long> accumulator, String
schemaJson) {
- add(accumulator, schemaJson, 1L);
+ public Map<String, Group> addInput(Map<String, Group> accumulator,
SchemaGroup file) {
+ add(accumulator, file.schemaJson, file.files, file.nullFreeColumns);
return accumulator;
}
@Override
- public Map<String, Long> mergeAccumulators(Iterable<Map<String, Long>>
accumulators) {
- Map<String, Long> merged = createAccumulator();
- for (Map<String, Long> accumulator : accumulators) {
- for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
- add(merged, entry.getKey(), entry.getValue());
+ public Map<String, Group> mergeAccumulators(Iterable<Map<String, Group>>
accumulators) {
+ Map<String, Group> merged = createAccumulator();
+ for (Map<String, Group> accumulator : accumulators) {
+ for (Map.Entry<String, Group> entry : accumulator.entrySet()) {
+ add(merged, entry.getKey(), entry.getValue().files,
entry.getValue().nullFreeColumns);
}
}
return merged;
}
@Override
- public List<KV<String, Long>> extractOutput(Map<String, Long> accumulator) {
- List<KV<String, Long>> schemas = new ArrayList<>();
- for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
- schemas.add(KV.of(entry.getKey(), entry.getValue()));
+ public List<SchemaGroup> extractOutput(Map<String, Group> accumulator) {
+ List<SchemaGroup> schemas = new ArrayList<>();
+ for (Map.Entry<String, Group> entry : accumulator.entrySet()) {
+ schemas.add(
+ new SchemaGroup(
+ entry.getKey(),
+ entry.getValue().files,
+ new ArrayList<>(entry.getValue().nullFreeColumns)));
}
schemas.sort(
(a, b) -> {
- int byCount = Long.compare(b.getValue(), a.getValue());
+ int byCount = Long.compare(b.files, a.files);
if (byCount != 0) {
return byCount;
}
- return a.getKey().compareTo(b.getKey());
+ return a.schemaJson.compareTo(b.schemaJson);
});
return schemas;
}
@Override
- public Coder<Map<String, Long>> getAccumulatorCoder(
- CoderRegistry registry, Coder<String> inputCoder) {
- return MapCoder.of(StringUtf8Coder.of(), VarLongCoder.of());
+ public Coder<Map<String, Group>> getAccumulatorCoder(
+ CoderRegistry registry, Coder<SchemaGroup> inputCoder) {
+ return MapCoder.of(StringUtf8Coder.of(), GroupCoder.INSTANCE);
}
@Override
- public Coder<List<KV<String, Long>>> getDefaultOutputCoder(
- CoderRegistry registry, Coder<String> inputCoder) {
- return ListCoder.of(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()));
+ public Coder<List<SchemaGroup>> getDefaultOutputCoder(
+ CoderRegistry registry, Coder<SchemaGroup> inputCoder) {
+ return outputCoder();
}
- private static void add(Map<String, Long> accumulator, String schemaJson,
long count) {
- Long existing = accumulator.get(schemaJson);
+ static Coder<SchemaGroup> groupCoder() {
+ return SchemaGroupCoder.INSTANCE;
+ }
+
+ static Coder<List<SchemaGroup>> outputCoder() {
+ return ListCoder.of(SchemaGroupCoder.INSTANCE);
+ }
+
+ private static final Coder<List<String>> COLUMNS_CODER =
ListCoder.of(StringUtf8Coder.of());
+
+ /** Singletons with class equality, so repeated mentions compare equal;
deterministic encoding. */
+ private static class GroupCoder extends CustomCoder<Group> {
Review Comment:
nit: use AtomicCoder to reduce boilerplate
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java:
##########
@@ -18,29 +18,208 @@
package org.apache.beam.sdk.io.iceberg;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
/**
- * Derives the schema a file contributes to schema inference. The canonical
form sorts struct fields
- * by name at every level and renumbers ids in deterministic order, so files
that differ only in
- * column order produce identical JSON. Ids are positional and meaningless:
the commit side
- * reconciles columns by name.
+ * What a file contributes to schema inference: the canonical form of the
schema it declares, and
+ * the columns its footer proves free of nulls.
+ *
+ * <p>The schema half depends only on the declared schema, never on the data,
so files written by
+ * the same job dedup to one entry no matter where their nulls fall. The null
evidence is combined
+ * per schema by {@link CollectDistinctSchemas} and reapplied by the commit
side via {@link
+ * #markRequired}.
+ *
+ * <p>The canonical form sorts struct fields by name at every level and
renumbers ids. Ids are
+ * positional and meaningless (the commit side reconciles columns by name), so
never diff two file
+ * schemas by id. Other field attributes (doc, defaults) are preserved,
matching what SchemaDelta
+ * compares. Column paths are dotted, like pins.
*/
final class FileSchemas {
private FileSchemas() {}
+ /** Canonical JSON of the schema the file declares. */
static String canonicalJson(ParquetMetadata footer) {
Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
return SchemaParser.toJson(canonical(converted));
}
+ /** This file as a schema group of one: its declared schema and its
null-free columns. */
+ static CollectDistinctSchemas.SchemaGroup schemaGroup(ParquetMetadata
footer) {
+ Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+ Schema tightened = tighten(converted, footer);
+ return new CollectDistinctSchemas.SchemaGroup(
+ SchemaParser.toJson(canonical(converted)), 1,
changedToRequired(converted, tightened));
+ }
+
+ /**
+ * Marks a declared-optional column required when every row group has a null
count of zero for it,
+ * so the file does not request a relaxation it does not need; an absent
count is not proof. A
+ * struct is null-free when any leaf under it is (a null struct nulls all
its leaves). Nothing
+ * under lists or maps is tightened: a zero count there would be valid
evidence too, but mapping
+ * physical chunk paths (writer-dependent names like {@code list.element},
{@code array}) onto the
+ * converted schema is not worth it. A file with no row groups has no rows
and proves every
+ * column, matching how the pin check treats empty files.
+ */
+ static Schema tighten(Schema schema, ParquetMetadata footer) {
+ if (footer.getBlocks().isEmpty()) {
+ return new Schema(tightenAll(schema.asStruct()).fields());
+ }
+ Set<List<String>> zeroNullLeaves = leafPathsWithZeroNullCounts(footer);
+ if (zeroNullLeaves.isEmpty()) {
+ return schema;
+ }
+ return new Schema(
+ tightenStruct(schema.asStruct(), new ArrayList<>(),
zeroNullLeaves).struct.fields());
+ }
+
+ /** With no rows, nothing can hold a null: every leaf and struct outside
lists and maps. */
+ private static Types.StructType tightenAll(Types.StructType struct) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ Type type = field.type();
+ if (type.isStructType()) {
+ fields.add(withOptionality(field, tightenAll(type.asStructType()),
false));
+ } else if (type.isPrimitiveType()) {
+ fields.add(withOptionality(field, type, false));
+ } else {
+ fields.add(field);
+ }
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /**
+ * Returns the schema with the given dotted column paths made required. The
commit side parses a
+ * group's schema JSON (optionality as the writer declared it) and applies
the group's null-free
+ * columns with this before classifying, so only relaxations some file
actually needs remain.
+ */
+ static Schema markRequired(Schema declared, Collection<String> columns) {
+ if (columns.isEmpty()) {
+ return declared;
+ }
+ Types.StructType required = markRequiredStruct(declared.asStruct(), "",
new HashSet<>(columns));
+ return new Schema(required.fields());
+ }
+
+ private static Types.StructType markRequiredStruct(
+ Types.StructType struct, String prefix, Set<String> columns) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ String path = prefix + field.name();
+ Type type = field.type();
+ if (type.isStructType()) {
+ type = markRequiredStruct(type.asStructType(), path + ".", columns);
+ }
+ boolean required = !field.isRequired() && columns.contains(path);
+ fields.add(withOptionality(field, type, field.isOptional() &&
!required));
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /** Dotted paths of fields the tightened schema made required, sorted. */
+ private static List<String> changedToRequired(Schema declared, Schema
tightened) {
+ List<String> paths = new ArrayList<>();
+ collectChangedToRequired(declared.asStruct(), tightened.asStruct(), "",
paths);
+ Collections.sort(paths);
+ return paths;
+ }
+
+ private static void collectChangedToRequired(
+ Types.StructType declared, Types.StructType tightened, String prefix,
List<String> out) {
+ for (int i = 0; i < declared.fields().size(); i++) {
+ Types.NestedField before = declared.fields().get(i);
+ Types.NestedField after = tightened.fields().get(i);
+ String path = prefix + before.name();
+ if (before.isOptional() && after.isRequired()) {
+ out.add(path);
+ }
+ if (before.type().isStructType()) {
+ collectChangedToRequired(
+ before.type().asStructType(), after.type().asStructType(), path +
".", out);
+ }
+ }
+ }
+
+ /** Leaf paths proven null-free in every row group (intersection over
blocks). */
+ private static Set<List<String>> leafPathsWithZeroNullCounts(ParquetMetadata
footer) {
+ Set<List<String>> proven = null;
+ for (BlockMetaData block : footer.getBlocks()) {
+ Set<List<String>> provenHere = new HashSet<>();
+ for (ColumnChunkMetaData chunk : block.getColumns()) {
+ // A zero-row row group proves trivially: zero rows hold zero nulls.
Review Comment:
this comment feels out of place
--
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]
