You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2020/05/14 18:00:49 UTC
[incubator-iceberg] branch master updated: Refactor ManifestReader
(#1036)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 55a4e57 Refactor ManifestReader (#1036)
55a4e57 is described below
commit 55a4e5725387ac618015cfa586e5c4e25ed8ba9a
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Thu May 14 11:00:43 2020 -0700
Refactor ManifestReader (#1036)
---
.../main/java/org/apache/iceberg/Filterable.java | 127 ---------
.../org/apache/iceberg/BaseManifestReader.java | 294 +++++++++++++++++++++
.../org/apache/iceberg/BaseRewriteManifests.java | 6 +-
.../main/java/org/apache/iceberg/BaseSnapshot.java | 1 -
.../java/org/apache/iceberg/FilteredManifest.java | 194 --------------
.../org/apache/iceberg/ManifestEntriesTable.java | 2 +-
.../java/org/apache/iceberg/ManifestGroup.java | 12 +-
.../java/org/apache/iceberg/ManifestReader.java | 220 +--------------
.../java/org/apache/iceberg/TableTestBase.java | 16 +-
.../org/apache/iceberg/TestManifestReader.java | 26 +-
10 files changed, 332 insertions(+), 566 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/Filterable.java b/api/src/main/java/org/apache/iceberg/Filterable.java
deleted file mode 100644
index d53a318..0000000
--- a/api/src/main/java/org/apache/iceberg/Filterable.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.iceberg;
-
-import com.google.common.collect.Lists;
-import java.util.Collection;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.io.CloseableIterable;
-
-/**
- * Methods to filter files in a snapshot or manifest when reading.
- *
- * @param <T> Java class returned by filter methods, also filterable
- */
-public interface Filterable<T extends Filterable<T>> extends CloseableIterable<DataFile> {
- /**
- * Selects the columns of a file manifest to read.
- * <p>
- * If columns are set multiple times, the last set of columns will be read.
- * <p>
- * If the Filterable object has partition filters, they will be added to the returned partial.
- * <p>
- * For a list of column names, see the table format specification.
- *
- * @param columns String column names to load from the manifest file
- * @return a Filterable that will load only the given columns
- */
- default T select(String... columns) {
- return select(Lists.newArrayList(columns));
- }
-
- /**
- * Selects the columns of a file manifest to read.
- * <p>
- * If columns are set multiple times, the last set of columns will be read.
- * <p>
- * If the Filterable object has partition filters, they will be added to the returned partial.
- * <p>
- * For a list of column names, see the table format specification.
- *
- * @param columns String column names to load from the manifest file
- * @return a Filterable that will load only the given columns
- */
- T select(Collection<String> columns);
-
- /**
- * Set the projection from a schema.
- *
- * @param fileProjection a projection of the DataFile schema
- * @return a Filterable that will load only the given schema's columns
- */
- T project(Schema fileProjection);
-
- /**
- * Adds a filter expression on partition data for matching files.
- * <p>
- * If the Filterable object already has partition filters, the new filter will be added as an
- * additional requirement. The result filter expression will be the result of expr and any
- * existing filters.
- * <p>
- * If the Filterable object has columns selected, they will be added to the returned partial.
- *
- * @param expr An expression for filtering this Filterable using partition data
- * @return a Filterable that will load only rows that match expr
- */
- T filterPartitions(Expression expr);
-
- /**
- * Adds a filter expression on data rows for matching files.
- * <p>
- * Expressions passed to this function will be converted to partition expressions before they are
- * used to filter data files.
- * <p>
- * If the Filterable object already has partition filters, the new filter will be added as an
- * additional requirement. The result filter expression will be the result of expr and any
- * existing filters.
- * <p>
- * If the Filterable object has columns selected, they will be added to the returned partial.
- *
- * @param expr An expression for filtering this Filterable using row data
- * @return a Filterable that will load only rows that match expr
- */
- T filterRows(Expression expr);
-
- /**
- * Sets case sensitivity.
- *
- * @param isCaseSensitive true if expression binding and schema projection should be case sensitive
- * @return a Filterable that will use the specified case sensitivity
- */
- T caseSensitive(boolean isCaseSensitive);
-
- /**
- * Sets case sensitive binding and projection.
- *
- * @return a Filterable that will case sensitive binding and projection
- */
- default T caseSensitive() {
- return caseSensitive(true);
- }
-
- /**
- * Sets case insensitive binding and projection.
- *
- * @return a Filterable that will case insensitive binding and projection
- */
- default T caseInsensitive() {
- return caseSensitive(false);
- }
-}
diff --git a/core/src/main/java/org/apache/iceberg/BaseManifestReader.java b/core/src/main/java/org/apache/iceberg/BaseManifestReader.java
new file mode 100644
index 0000000..1496f01
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/BaseManifestReader.java
@@ -0,0 +1,294 @@
+/*
+ * 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.iceberg;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.avro.AvroIterable;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.types.Types;
+
+import static org.apache.iceberg.expressions.Expressions.alwaysTrue;
+
+/**
+ * Base reader for data and delete manifest files.
+ *
+ * @param <T> The Java class of files returned by this reader.
+ * @param <ThisT> The Java class of this reader, returned by configuration methods.
+ */
+abstract class BaseManifestReader<T, ThisT> extends CloseableGroup implements CloseableIterable<T> {
+ static final ImmutableList<String> ALL_COLUMNS = ImmutableList.of("*");
+ private static final Set<String> STATS_COLUMNS = Sets.newHashSet(
+ "value_counts", "null_value_counts", "lower_bounds", "upper_bounds");
+
+ protected enum FileType {
+ DATA_FILES(GenericDataFile.class.getName()),
+ DELETE_FILES("...");
+
+ private final String fileClass;
+
+ FileType(String fileClass) {
+ this.fileClass = fileClass;
+ }
+
+ private String fileClass() {
+ return fileClass;
+ }
+ }
+
+ private final InputFile file;
+ private final InheritableMetadata inheritableMetadata;
+ private final FileType content;
+ private final Map<String, String> metadata;
+ private final PartitionSpec spec;
+ private final Schema fileSchema;
+
+ // updated by configuration methods
+ private Expression partFilter = alwaysTrue();
+ private Expression rowFilter = alwaysTrue();
+ private Schema fileProjection = null;
+ private Collection<String> columns = null;
+ private boolean caseSensitive = true;
+
+ // lazily initialized
+ private Evaluator lazyEvaluator = null;
+ private InclusiveMetricsEvaluator lazyMetricsEvaluator = null;
+
+ protected BaseManifestReader(InputFile file, Map<Integer, PartitionSpec> specsById,
+ InheritableMetadata inheritableMetadata, FileType content) {
+ this.file = file;
+ this.inheritableMetadata = inheritableMetadata;
+ this.content = content;
+
+ try {
+ try (AvroIterable<ManifestEntry> headerReader = Avro.read(file)
+ .project(ManifestEntry.getSchema(Types.StructType.of()).select("status"))
+ .build()) {
+ this.metadata = headerReader.getMetadata();
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e);
+ }
+
+ int specId = TableMetadata.INITIAL_SPEC_ID;
+ String specProperty = metadata.get("partition-spec-id");
+ if (specProperty != null) {
+ specId = Integer.parseInt(specProperty);
+ }
+
+ if (specsById != null) {
+ this.spec = specsById.get(specId);
+ } else {
+ Schema schema = SchemaParser.fromJson(metadata.get("schema"));
+ this.spec = PartitionSpecParser.fromJsonFields(schema, specId, metadata.get("partition-spec"));
+ }
+
+ this.fileSchema = new Schema(DataFile.getType(spec.partitionType()).fields());
+ }
+
+ protected abstract ThisT self();
+
+ public InputFile file() {
+ return file;
+ }
+
+ public Schema schema() {
+ return fileSchema;
+ }
+
+ public PartitionSpec spec() {
+ return spec;
+ }
+
+ public ThisT select(Collection<String> newColumns) {
+ Preconditions.checkState(fileProjection == null,
+ "Cannot select columns using both select(String...) and project(Schema)");
+ this.columns = newColumns;
+ return self();
+ }
+
+ public ThisT project(Schema newFileProjection) {
+ Preconditions.checkState(columns == null,
+ "Cannot select columns using both select(String...) and project(Schema)");
+ this.fileProjection = newFileProjection;
+ return self();
+ }
+
+ public ThisT filterPartitions(Expression expr) {
+ this.partFilter = Expressions.and(partFilter, expr);
+ return self();
+ }
+
+ public ThisT filterRows(Expression expr) {
+ this.rowFilter = Expressions.and(rowFilter, expr);
+ return self();
+ }
+
+ public ThisT caseSensitive(boolean isCaseSensitive) {
+ this.caseSensitive = isCaseSensitive;
+ return self();
+ }
+
+ CloseableIterable<ManifestEntry> entries() {
+ if ((rowFilter != null && rowFilter != Expressions.alwaysTrue()) ||
+ (partFilter != null && partFilter != Expressions.alwaysTrue())) {
+ Evaluator evaluator = evaluator();
+ InclusiveMetricsEvaluator metricsEvaluator = metricsEvaluator();
+
+ // ensure stats columns are present for metrics evaluation
+ boolean requireStatsProjection = requireStatsProjection(rowFilter, columns);
+ Collection<String> projectColumns = requireStatsProjection ? withStatsColumns(columns) : columns;
+
+ return CloseableIterable.filter(
+ open(projection(fileSchema, fileProjection, projectColumns, caseSensitive)),
+ entry -> entry != null &&
+ evaluator.eval(entry.file().partition()) &&
+ metricsEvaluator.eval(entry.file()));
+ } else {
+ return open(projection(fileSchema, fileProjection, columns, caseSensitive));
+ }
+ }
+
+ private CloseableIterable<ManifestEntry> open(Schema projection) {
+ FileFormat format = FileFormat.fromFileName(file.location());
+ Preconditions.checkArgument(format != null, "Unable to determine format of manifest: %s", file);
+
+ switch (format) {
+ case AVRO:
+ AvroIterable<ManifestEntry> reader = Avro.read(file)
+ .project(ManifestEntry.wrapFileSchema(projection.asStruct()))
+ .rename("manifest_entry", GenericManifestEntry.class.getName())
+ .rename("partition", PartitionData.class.getName())
+ .rename("r102", PartitionData.class.getName())
+ .rename("data_file", content.fileClass())
+ .rename("r2", content.fileClass())
+ .classLoader(GenericManifestEntry.class.getClassLoader())
+ .reuseContainers()
+ .build();
+
+ addCloseable(reader);
+
+ return CloseableIterable.transform(reader, inheritableMetadata::apply);
+
+ default:
+ throw new UnsupportedOperationException("Invalid format for manifest file: " + format);
+ }
+ }
+
+ CloseableIterable<ManifestEntry> liveEntries() {
+ return CloseableIterable.filter(entries(),
+ entry -> entry != null && entry.status() != ManifestEntry.Status.DELETED);
+ }
+
+ /**
+ * @return an Iterator of DataFile. Makes defensive copies of files before returning
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public Iterator<T> iterator() {
+ if (dropStats(rowFilter, columns)) {
+ return (Iterator<T>) CloseableIterable.transform(liveEntries(), e -> e.file().copyWithoutStats()).iterator();
+ } else {
+ return (Iterator<T>) CloseableIterable.transform(liveEntries(), e -> e.file().copy()).iterator();
+ }
+ }
+
+ private static Schema projection(Schema schema, Schema project, Collection<String> columns, boolean caseSensitive) {
+ if (columns != null) {
+ if (caseSensitive) {
+ return schema.select(columns);
+ } else {
+ return schema.caseInsensitiveSelect(columns);
+ }
+ } else if (project != null) {
+ return project;
+ }
+
+ return schema;
+ }
+
+ private Evaluator evaluator() {
+ if (lazyEvaluator == null) {
+ Expression projected = Projections.inclusive(spec, caseSensitive).project(rowFilter);
+ Expression finalPartFilter = Expressions.and(projected, partFilter);
+ if (finalPartFilter != null) {
+ this.lazyEvaluator = new Evaluator(spec.partitionType(), finalPartFilter, caseSensitive);
+ } else {
+ this.lazyEvaluator = new Evaluator(spec.partitionType(), Expressions.alwaysTrue(), caseSensitive);
+ }
+ }
+ return lazyEvaluator;
+ }
+
+ private InclusiveMetricsEvaluator metricsEvaluator() {
+ if (lazyMetricsEvaluator == null) {
+ if (rowFilter != null) {
+ this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
+ spec.schema(), rowFilter, caseSensitive);
+ } else {
+ this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
+ spec.schema(), Expressions.alwaysTrue(), caseSensitive);
+ }
+ }
+ return lazyMetricsEvaluator;
+ }
+
+ private static boolean requireStatsProjection(Expression rowFilter, Collection<String> columns) {
+ // Make sure we have all stats columns for metrics evaluator
+ return rowFilter != Expressions.alwaysTrue() &&
+ !columns.containsAll(BaseManifestReader.ALL_COLUMNS) &&
+ !columns.containsAll(STATS_COLUMNS);
+ }
+
+ static boolean dropStats(Expression rowFilter, Collection<String> columns) {
+ // Make sure we only drop all stats if we had projected all stats
+ // We do not drop stats even if we had partially added some stats columns
+ return rowFilter != Expressions.alwaysTrue() &&
+ !columns.containsAll(BaseManifestReader.ALL_COLUMNS) &&
+ Sets.intersection(Sets.newHashSet(columns), STATS_COLUMNS).isEmpty();
+ }
+
+ private static Collection<String> withStatsColumns(Collection<String> columns) {
+ if (columns.containsAll(BaseManifestReader.ALL_COLUMNS)) {
+ return columns;
+ } else {
+ List<String> projectColumns = Lists.newArrayList(columns);
+ projectColumns.addAll(STATS_COLUMNS); // order doesn't matter
+ return projectColumns;
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
index 5e1d3a2..2f8c5df 100644
--- a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
+++ b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
@@ -234,9 +234,9 @@ public class BaseRewriteManifests extends SnapshotProducer<RewriteManifests> imp
keptManifests.add(manifest);
} else {
rewrittenManifests.add(manifest);
- try (ManifestReader reader = ManifestFiles.read(manifest, ops.io(), ops.current().specsById())) {
- FilteredManifest filteredManifest = reader.select(Arrays.asList("*"));
- filteredManifest.liveEntries().forEach(
+ try (ManifestReader reader = ManifestFiles.read(manifest, ops.io(), ops.current().specsById())
+ .select(Arrays.asList("*"))) {
+ reader.liveEntries().forEach(
entry -> appendEntry(entry, clusterByFunc.apply(entry.file()), manifest.partitionSpecId())
);
diff --git a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java
index 22ab7fe..59f49f0 100644
--- a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java
+++ b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java
@@ -160,7 +160,6 @@ class BaseSnapshot implements Snapshot {
manifest -> Objects.equal(manifest.snapshotId(), snapshotId));
try (CloseableIterable<ManifestEntry> entries = new ManifestGroup(io, changedManifests)
.ignoreExisting()
- .select(ManifestReader.ALL_COLUMNS)
.entries()) {
for (ManifestEntry entry : entries) {
switch (entry.status()) {
diff --git a/core/src/main/java/org/apache/iceberg/FilteredManifest.java b/core/src/main/java/org/apache/iceberg/FilteredManifest.java
deleted file mode 100644
index 0018c0e..0000000
--- a/core/src/main/java/org/apache/iceberg/FilteredManifest.java
+++ /dev/null
@@ -1,194 +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.iceberg;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Set;
-import org.apache.iceberg.ManifestEntry.Status;
-import org.apache.iceberg.expressions.Evaluator;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
-import org.apache.iceberg.expressions.Projections;
-import org.apache.iceberg.io.CloseableIterable;
-
-public class FilteredManifest implements Filterable<FilteredManifest> {
- private static final Set<String> STATS_COLUMNS = Sets.newHashSet(
- "value_counts", "null_value_counts", "lower_bounds", "upper_bounds");
-
- private final ManifestReader reader;
- private final Expression partFilter;
- private final Expression rowFilter;
- private final Schema fileSchema;
- private final Collection<String> columns;
- private final boolean caseSensitive;
-
- // lazy state
- private Evaluator lazyEvaluator = null;
- private InclusiveMetricsEvaluator lazyMetricsEvaluator = null;
-
- FilteredManifest(ManifestReader reader, Expression partFilter, Expression rowFilter,
- Schema fileSchema, Collection<String> columns, boolean caseSensitive) {
- Preconditions.checkNotNull(reader, "ManifestReader cannot be null");
- this.reader = reader;
- this.partFilter = partFilter;
- this.rowFilter = rowFilter;
- this.fileSchema = fileSchema;
- this.columns = columns;
- this.caseSensitive = caseSensitive;
- }
-
- @Override
- public FilteredManifest select(Collection<String> selectedColumns) {
- return new FilteredManifest(reader, partFilter, rowFilter, fileSchema, selectedColumns, caseSensitive);
- }
-
- @Override
- public FilteredManifest project(Schema fileProjection) {
- return new FilteredManifest(reader, partFilter, rowFilter, fileProjection, columns, caseSensitive);
- }
-
- @Override
- public FilteredManifest filterPartitions(Expression expr) {
- return new FilteredManifest(
- reader, Expressions.and(partFilter, expr), rowFilter, fileSchema, columns, caseSensitive);
- }
-
- @Override
- public FilteredManifest filterRows(Expression expr) {
- return new FilteredManifest(
- reader, partFilter, Expressions.and(rowFilter, expr), fileSchema, columns, caseSensitive);
- }
-
- @Override
- public FilteredManifest caseSensitive(boolean isCaseSensitive) {
- return new FilteredManifest(reader, partFilter, rowFilter, fileSchema, columns, isCaseSensitive);
- }
-
- CloseableIterable<ManifestEntry> allEntries() {
- if ((rowFilter != null && rowFilter != Expressions.alwaysTrue()) ||
- (partFilter != null && partFilter != Expressions.alwaysTrue())) {
- Evaluator evaluator = evaluator();
- InclusiveMetricsEvaluator metricsEvaluator = metricsEvaluator();
-
- // ensure stats columns are present for metrics evaluation
- boolean requireStatsProjection = requireStatsProjection(rowFilter, columns);
- Collection<String> projectColumns = requireStatsProjection ? withStatsColumns(columns) : columns;
-
- return CloseableIterable.filter(
- reader.entries(projection(fileSchema, projectColumns, caseSensitive)),
- entry -> entry != null &&
- evaluator.eval(entry.file().partition()) &&
- metricsEvaluator.eval(entry.file()));
- } else {
- return reader.entries(projection(fileSchema, columns, caseSensitive));
- }
- }
-
- CloseableIterable<ManifestEntry> liveEntries() {
- return CloseableIterable.filter(allEntries(), entry -> entry != null && entry.status() != Status.DELETED);
- }
-
- /**
- * @return an Iterator of DataFile. Makes defensive copies of files before returning
- */
- @Override
- public Iterator<DataFile> iterator() {
- if (dropStats(rowFilter, columns)) {
- return CloseableIterable.transform(liveEntries(), e -> e.file().copyWithoutStats()).iterator();
- } else {
- return CloseableIterable.transform(liveEntries(), e -> e.file().copy()).iterator();
- }
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
- }
-
- private static Schema projection(Schema fileSchema, Collection<String> columns, boolean caseSensitive) {
- if (columns != null) {
- if (caseSensitive) {
- return fileSchema.select(columns);
- } else {
- return fileSchema.caseInsensitiveSelect(columns);
- }
- }
-
- return fileSchema;
- }
-
- private Evaluator evaluator() {
- if (lazyEvaluator == null) {
- Expression projected = Projections.inclusive(reader.spec(), caseSensitive).project(rowFilter);
- Expression finalPartFilter = Expressions.and(projected, partFilter);
- if (finalPartFilter != null) {
- this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), finalPartFilter, caseSensitive);
- } else {
- this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), Expressions.alwaysTrue(), caseSensitive);
- }
- }
- return lazyEvaluator;
- }
-
- private InclusiveMetricsEvaluator metricsEvaluator() {
- if (lazyMetricsEvaluator == null) {
- if (rowFilter != null) {
- this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
- reader.spec().schema(), rowFilter, caseSensitive);
- } else {
- this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
- reader.spec().schema(), Expressions.alwaysTrue(), caseSensitive);
- }
- }
- return lazyMetricsEvaluator;
- }
-
- static boolean requireStatsProjection(Expression rowFilter, Collection<String> columns) {
- // Make sure we have all stats columns for metrics evaluator
- return rowFilter != Expressions.alwaysTrue() &&
- !columns.containsAll(ManifestReader.ALL_COLUMNS) &&
- !columns.containsAll(STATS_COLUMNS);
- }
-
- static boolean dropStats(Expression rowFilter, Collection<String> columns) {
- // Make sure we only drop all stats if we had projected all stats
- // We do not drop stats even if we had partially added some stats columns
- return rowFilter != Expressions.alwaysTrue() &&
- !columns.containsAll(ManifestReader.ALL_COLUMNS) &&
- Sets.intersection(Sets.newHashSet(columns), STATS_COLUMNS).isEmpty();
- }
-
- private static Collection<String> withStatsColumns(Collection<String> columns) {
- if (columns.containsAll(ManifestReader.ALL_COLUMNS)) {
- return columns;
- } else {
- List<String> projectColumns = Lists.newArrayList(columns);
- projectColumns.addAll(STATS_COLUMNS); // order doesn't matter
- return projectColumns;
- }
- }
-}
diff --git a/core/src/main/java/org/apache/iceberg/ManifestEntriesTable.java b/core/src/main/java/org/apache/iceberg/ManifestEntriesTable.java
index d2458f8..4e00549 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestEntriesTable.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestEntriesTable.java
@@ -128,7 +128,7 @@ public class ManifestEntriesTable extends BaseMetadataTable {
@Override
public CloseableIterable<StructLike> rows() {
return CloseableIterable.transform(
- ManifestFiles.read(manifest, io).project(fileSchema).allEntries(),
+ ManifestFiles.read(manifest, io).project(fileSchema).entries(),
file -> (GenericManifestEntry) file);
}
diff --git a/core/src/main/java/org/apache/iceberg/ManifestGroup.java b/core/src/main/java/org/apache/iceberg/ManifestGroup.java
index 767fae7..ac7f146 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestGroup.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestGroup.java
@@ -66,7 +66,7 @@ class ManifestGroup {
this.partitionFilter = Expressions.alwaysTrue();
this.ignoreDeleted = false;
this.ignoreExisting = false;
- this.columns = ManifestReader.ALL_COLUMNS;
+ this.columns = BaseManifestReader.ALL_COLUMNS;
this.caseSensitive = true;
this.manifestPredicate = m -> true;
this.manifestEntryPredicate = e -> true;
@@ -138,7 +138,7 @@ class ManifestGroup {
PartitionSpec spec = specsById.get(specId);
return ResidualEvaluator.of(spec, dataFilter, caseSensitive);
});
- boolean dropStats = FilteredManifest.dropStats(dataFilter, columns);
+ boolean dropStats = BaseManifestReader.dropStats(dataFilter, columns);
Iterable<CloseableIterable<FileScanTask>> tasks = entries((manifest, entries) -> {
int partitionSpecId = manifest.partitionSpecId();
PartitionSpec spec = specsById.get(partitionSpecId);
@@ -209,17 +209,15 @@ class ManifestGroup {
Iterable<CloseableIterable<T>> readers = Iterables.transform(
matchingManifests,
manifest -> {
- ManifestReader reader = ManifestFiles.read(manifest, io, specsById);
-
- FilteredManifest filtered = reader
+ ManifestReader reader = ManifestFiles.read(manifest, io, specsById)
.filterRows(dataFilter)
.filterPartitions(partitionFilter)
.caseSensitive(caseSensitive)
.select(columns);
- CloseableIterable<ManifestEntry> entries = filtered.allEntries();
+ CloseableIterable<ManifestEntry> entries = reader.entries();
if (ignoreDeleted) {
- entries = filtered.liveEntries();
+ entries = reader.liveEntries();
}
if (ignoreExisting) {
diff --git a/core/src/main/java/org/apache/iceberg/ManifestReader.java b/core/src/main/java/org/apache/iceberg/ManifestReader.java
index 045ebe1..45c19fc 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestReader.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestReader.java
@@ -19,232 +19,22 @@
package org.apache.iceberg;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.function.Function;
-import org.apache.iceberg.avro.Avro;
-import org.apache.iceberg.avro.AvroIterable;
-import org.apache.iceberg.exceptions.RuntimeIOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.io.CloseableGroup;
-import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
-import org.apache.iceberg.types.Types;
-
-import static org.apache.iceberg.expressions.Expressions.alwaysTrue;
/**
* Reader for manifest files.
* <p>
* Create readers using {@link ManifestFiles#read(ManifestFile, FileIO, Map)}.
*/
-public class ManifestReader extends CloseableGroup implements Filterable<FilteredManifest> {
- static final ImmutableList<String> ALL_COLUMNS = ImmutableList.of("*");
- static final ImmutableList<String> CHANGE_COLUMNS = ImmutableList.of(
- "file_path", "file_format", "partition", "record_count", "file_size_in_bytes");
- static final ImmutableList<String> CHANGE_WITH_STATS_COLUMNS = ImmutableList.<String>builder()
- .addAll(CHANGE_COLUMNS)
- .add("value_counts", "null_value_counts", "lower_bounds", "upper_bounds")
- .build();
-
- /**
- * Returns a new {@link ManifestReader} for an {@link InputFile}.
- * <p>
- * <em>Note:</em> Callers should use {@link ManifestFiles#read(ManifestFile, FileIO, Map)} to ensure that
- * manifest entries with partial metadata can inherit missing properties from the manifest metadata.
- *
- * @param file an InputFile
- * @return a manifest reader
- * @deprecated will be removed in 0.9.0; use {@link ManifestFiles#read(ManifestFile, FileIO, Map)} instead.
- */
- @Deprecated
- public static ManifestReader read(InputFile file) {
- return read(file, null);
- }
-
- /**
- * Returns a new {@link ManifestReader} for an {@link InputFile}.
- *
- * @param file an InputFile
- * @param specLookup a function to look up the manifest's partition spec by ID
- * @return a manifest reader
- * @deprecated will be removed in 0.9.0; use {@link ManifestFiles#read(ManifestFile, FileIO, Map)} instead.
- */
- @Deprecated
- public static ManifestReader read(InputFile file, Function<Integer, PartitionSpec> specLookup) {
- return new ManifestReader(file, specLookup, InheritableMetadataFactory.empty());
- }
-
- private final InputFile file;
- private final InheritableMetadata inheritableMetadata;
- private final Map<String, String> metadata;
- private final PartitionSpec spec;
- private final Schema fileSchema;
-
- // lazily initialized
- private List<ManifestEntry> cachedAdds = null;
- private List<ManifestEntry> cachedDeletes = null;
-
+public class ManifestReader extends BaseManifestReader<DataFile, ManifestReader> {
ManifestReader(InputFile file, Map<Integer, PartitionSpec> specsById,
- InheritableMetadata inheritableMetadata) {
- this(file, specsById != null ? specsById::get : null, inheritableMetadata);
+ InheritableMetadata inheritableMetadata) {
+ super(file, specsById, inheritableMetadata, FileType.DATA_FILES);
}
- private ManifestReader(InputFile file, Function<Integer, PartitionSpec> specLookup,
- InheritableMetadata inheritableMetadata) {
- this.file = file;
- this.inheritableMetadata = inheritableMetadata;
-
- try {
- try (AvroIterable<ManifestEntry> headerReader = Avro.read(file)
- .project(ManifestEntry.getSchema(Types.StructType.of()).select("status"))
- .build()) {
- this.metadata = headerReader.getMetadata();
- }
- } catch (IOException e) {
- throw new RuntimeIOException(e);
- }
-
- int specId = TableMetadata.INITIAL_SPEC_ID;
- String specProperty = metadata.get("partition-spec-id");
- if (specProperty != null) {
- specId = Integer.parseInt(specProperty);
- }
-
- if (specLookup != null) {
- this.spec = specLookup.apply(specId);
- } else {
- Schema schema = SchemaParser.fromJson(metadata.get("schema"));
- this.spec = PartitionSpecParser.fromJsonFields(schema, specId, metadata.get("partition-spec"));
- }
-
- this.fileSchema = new Schema(DataFile.getType(spec.partitionType()).fields());
+ protected ManifestReader self() {
+ return this;
}
-
- public InputFile file() {
- return file;
- }
-
- public Schema schema() {
- return fileSchema;
- }
-
- public PartitionSpec spec() {
- return spec;
- }
-
- @Override
- public FilteredManifest select(Collection<String> columns) {
- return new FilteredManifest(this, alwaysTrue(), alwaysTrue(), fileSchema, columns, true);
- }
-
- @Override
- public FilteredManifest project(Schema fileProjection) {
- return new FilteredManifest(this, alwaysTrue(), alwaysTrue(), fileProjection, ALL_COLUMNS, true);
- }
-
- @Override
- public FilteredManifest filterPartitions(Expression expr) {
- return new FilteredManifest(this, expr, alwaysTrue(), fileSchema, ALL_COLUMNS, true);
- }
-
- @Override
- public FilteredManifest filterRows(Expression expr) {
- return new FilteredManifest(this, alwaysTrue(), expr, fileSchema, ALL_COLUMNS, true);
- }
-
- @Override
- public FilteredManifest caseSensitive(boolean caseSensitive) {
- return new FilteredManifest(this, alwaysTrue(), alwaysTrue(), fileSchema, ALL_COLUMNS, caseSensitive);
- }
-
- public List<ManifestEntry> addedFiles() {
- if (cachedAdds == null) {
- cacheChanges();
- }
- return cachedAdds;
- }
-
- public List<ManifestEntry> deletedFiles() {
- if (cachedDeletes == null) {
- cacheChanges();
- }
- return cachedDeletes;
- }
-
- private void cacheChanges() {
- List<ManifestEntry> adds = Lists.newArrayList();
- List<ManifestEntry> deletes = Lists.newArrayList();
-
- try (CloseableIterable<ManifestEntry> entries = entries(fileSchema.select(CHANGE_COLUMNS))) {
- for (ManifestEntry entry : entries) {
- switch (entry.status()) {
- case ADDED:
- adds.add(entry.copyWithoutStats());
- break;
- case DELETED:
- deletes.add(entry.copyWithoutStats());
- break;
- default:
- }
- }
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to close manifest entries");
- }
-
- this.cachedAdds = adds;
- this.cachedDeletes = deletes;
- }
-
- CloseableIterable<ManifestEntry> entries() {
- return entries(fileSchema);
- }
-
- CloseableIterable<ManifestEntry> entries(Schema fileProjection) {
- FileFormat format = FileFormat.fromFileName(file.location());
- Preconditions.checkArgument(format != null, "Unable to determine format of manifest: %s", file);
-
- switch (format) {
- case AVRO:
- AvroIterable<ManifestEntry> reader = Avro.read(file)
- .project(ManifestEntry.wrapFileSchema(fileProjection.asStruct()))
- .rename("manifest_entry", GenericManifestEntry.class.getName())
- .rename("partition", PartitionData.class.getName())
- .rename("r102", PartitionData.class.getName())
- .rename("data_file", GenericDataFile.class.getName())
- .rename("r2", GenericDataFile.class.getName())
- .classLoader(GenericManifestFile.class.getClassLoader())
- .reuseContainers()
- .build();
-
- addCloseable(reader);
-
- return CloseableIterable.transform(reader, inheritableMetadata::apply);
-
- default:
- throw new UnsupportedOperationException("Invalid format for manifest file: " + format);
- }
- }
-
- @Override
- public Iterator<DataFile> iterator() {
- return iterator(fileSchema);
- }
-
- // visible for use by PartialManifest
- Iterator<DataFile> iterator(Schema fileProjection) {
- return Iterables.transform(Iterables.filter(
- entries(fileProjection),
- entry -> entry.status() != ManifestEntry.Status.DELETED),
- ManifestEntry::file).iterator();
- }
-
}
diff --git a/core/src/test/java/org/apache/iceberg/TableTestBase.java b/core/src/test/java/org/apache/iceberg/TableTestBase.java
index ef4466d..7e4b895 100644
--- a/core/src/test/java/org/apache/iceberg/TableTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/TableTestBase.java
@@ -132,11 +132,15 @@ public class TableTestBase {
}
ManifestFile writeManifest(DataFile... files) throws IOException {
+ return writeManifest(null, files);
+ }
+
+ ManifestFile writeManifest(Long snapshotId, DataFile... files) throws IOException {
File manifestFile = temp.newFile("input.m0.avro");
Assert.assertTrue(manifestFile.delete());
OutputFile outputFile = table.ops().io().newOutputFile(manifestFile.getCanonicalPath());
- ManifestWriter writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, null);
+ ManifestWriter writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, snapshotId);
try {
for (DataFile file : files) {
writer.add(file);
@@ -149,11 +153,19 @@ public class TableTestBase {
}
ManifestFile writeManifest(String fileName, ManifestEntry... entries) throws IOException {
+ return writeManifest(null, fileName, entries);
+ }
+
+ ManifestFile writeManifest(Long snapshotId, ManifestEntry... entries) throws IOException {
+ return writeManifest(snapshotId, "input.m0.avro", entries);
+ }
+
+ ManifestFile writeManifest(Long snapshotId, String fileName, ManifestEntry... entries) throws IOException {
File manifestFile = temp.newFile(fileName);
Assert.assertTrue(manifestFile.delete());
OutputFile outputFile = table.ops().io().newOutputFile(manifestFile.getCanonicalPath());
- ManifestWriter writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, null);
+ ManifestWriter writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, snapshotId);
try {
for (ManifestEntry entry : entries) {
writer.addEntry(entry);
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestReader.java b/core/src/test/java/org/apache/iceberg/TestManifestReader.java
index 058a845..883d341 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestReader.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestReader.java
@@ -44,10 +44,9 @@ public class TestManifestReader extends TableTestBase {
}
@Test
- @SuppressWarnings("deprecation")
public void testManifestReaderWithEmptyInheritableMetadata() throws IOException {
- ManifestFile manifest = writeManifest("manifest.avro", manifestEntry(Status.EXISTING, 1000L, FILE_A));
- try (ManifestReader reader = ManifestReader.read(FILE_IO.newInputFile(manifest.path()))) {
+ ManifestFile manifest = writeManifest(1000L, manifestEntry(Status.EXISTING, 1000L, FILE_A));
+ try (ManifestReader reader = ManifestFiles.read(manifest, FILE_IO)) {
ManifestEntry entry = Iterables.getOnlyElement(reader.entries());
Assert.assertEquals(Status.EXISTING, entry.status());
Assert.assertEquals(FILE_A.path(), entry.file().path());
@@ -56,22 +55,18 @@ public class TestManifestReader extends TableTestBase {
}
@Test
- @SuppressWarnings("deprecation")
public void testInvalidUsage() throws IOException {
ManifestFile manifest = writeManifest(FILE_A, FILE_B);
- try (ManifestReader reader = ManifestReader.read(FILE_IO.newInputFile(manifest.path()))) {
- AssertHelpers.assertThrows(
- "Should not be possible to read manifest without explicit snapshot ids and inheritable metadata",
- IllegalArgumentException.class, "must have explicit snapshot ids",
- () -> Iterables.getOnlyElement(reader.entries()));
- }
+ AssertHelpers.assertThrows(
+ "Should not be possible to read manifest without explicit snapshot ids and inheritable metadata",
+ IllegalArgumentException.class, "Cannot read from ManifestFile with null (unassigned) snapshot ID",
+ () -> ManifestFiles.read(manifest, FILE_IO));
}
@Test
- @SuppressWarnings("deprecation")
public void testManifestReaderWithPartitionMetadata() throws IOException {
- ManifestFile manifest = writeManifest("manifest.avro", manifestEntry(Status.EXISTING, 123L, FILE_A));
- try (ManifestReader reader = ManifestReader.read(FILE_IO.newInputFile(manifest.path()))) {
+ ManifestFile manifest = writeManifest(1000L, manifestEntry(Status.EXISTING, 123L, FILE_A));
+ try (ManifestReader reader = ManifestFiles.read(manifest, FILE_IO)) {
ManifestEntry entry = Iterables.getOnlyElement(reader.entries());
Assert.assertEquals(123L, (long) entry.snapshotId());
@@ -84,7 +79,6 @@ public class TestManifestReader extends TableTestBase {
}
@Test
- @SuppressWarnings("deprecation")
public void testManifestReaderWithUpdatedPartitionMetadataForV1Table() throws IOException {
PartitionSpec spec = PartitionSpec.builderFor(table.schema())
.bucket("id", 8)
@@ -92,8 +86,8 @@ public class TestManifestReader extends TableTestBase {
.build();
table.ops().commit(table.ops().current(), table.ops().current().updatePartitionSpec(spec));
- ManifestFile manifest = writeManifest("manifest.avro", manifestEntry(Status.EXISTING, 123L, FILE_A));
- try (ManifestReader reader = ManifestReader.read(FILE_IO.newInputFile(manifest.path()))) {
+ ManifestFile manifest = writeManifest(1000L, manifestEntry(Status.EXISTING, 123L, FILE_A));
+ try (ManifestReader reader = ManifestFiles.read(manifest, FILE_IO)) {
ManifestEntry entry = Iterables.getOnlyElement(reader.entries());
Assert.assertEquals(123L, (long) entry.snapshotId());