-
Notifications
You must be signed in to change notification settings - Fork 3.4k
Core: Add all_delete_files and all_files tables #4694
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * 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 org.apache.iceberg.io.CloseableIterable; | ||
|
|
||
| /** | ||
| * A {@link Table} implementation that exposes a table's valid delete files as rows. | ||
| * <p> | ||
| * A valid delete file is one that is readable from any snapshot currently tracked by the table. | ||
| * <p> | ||
| * This table may return duplicate rows. | ||
| */ | ||
| public class AllDeleteFilesTable extends BaseFilesTable { | ||
|
|
||
| AllDeleteFilesTable(TableOperations ops, Table table) { | ||
| this(ops, table, table.name() + ".all_delete_files"); | ||
| } | ||
|
|
||
| AllDeleteFilesTable(TableOperations ops, Table table, String name) { | ||
| super(ops, table, name); | ||
| } | ||
|
|
||
| @Override | ||
| public TableScan newScan() { | ||
| return new AllDataFilesTableScan(operations(), table(), schema()); | ||
| } | ||
|
|
||
| @Override | ||
| MetadataTableType metadataTableType() { | ||
| return MetadataTableType.ALL_DELETE_FILES; | ||
| } | ||
|
|
||
| public static class AllDataFilesTableScan extends BaseAllFilesTableScan { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Typo: should be
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah good catch .. |
||
|
|
||
| AllDataFilesTableScan(TableOperations ops, Table table, Schema schema) { | ||
| super(ops, table, schema, MetadataTableType.ALL_DELETE_FILES); | ||
| } | ||
|
|
||
| private AllDataFilesTableScan(TableOperations ops, Table table, Schema schema, | ||
| TableScanContext context) { | ||
| super(ops, table, schema, MetadataTableType.ALL_DELETE_FILES, context); | ||
| } | ||
|
|
||
| @Override | ||
| protected TableScan newRefinedScan(TableOperations ops, Table table, Schema schema, TableScanContext context) { | ||
| return new AllDataFilesTableScan(ops, table, schema, context); | ||
| } | ||
|
|
||
| @Override | ||
| protected CloseableIterable<ManifestFile> manifests() { | ||
| return reachableManifests(Snapshot::deleteManifests); | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * 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 org.apache.iceberg.io.CloseableIterable; | ||
|
|
||
| /** | ||
| * A {@link Table} implementation that exposes a table's valid files as rows. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: also redundant |
||
| * <p> | ||
| * A valid file is one that is readable from any snapshot currently tracked by the table. | ||
| * <p> | ||
| * This table may return duplicate rows. | ||
| */ | ||
| public class AllFilesTable extends BaseFilesTable { | ||
|
|
||
| AllFilesTable(TableOperations ops, Table table) { | ||
| this(ops, table, table.name() + ".all_files"); | ||
| } | ||
|
|
||
| AllFilesTable(TableOperations ops, Table table, String name) { | ||
| super(ops, table, name); | ||
| } | ||
|
|
||
| @Override | ||
| public TableScan newScan() { | ||
| return new AllDataFilesTableScan(operations(), table(), schema()); | ||
| } | ||
|
|
||
| @Override | ||
| MetadataTableType metadataTableType() { | ||
| return MetadataTableType.ALL_FILES; | ||
| } | ||
|
|
||
| public static class AllDataFilesTableScan extends BaseAllFilesTableScan { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Typo in the name as well. |
||
|
|
||
| AllDataFilesTableScan(TableOperations ops, Table table, Schema schema) { | ||
| super(ops, table, schema, MetadataTableType.ALL_FILES); | ||
| } | ||
|
|
||
| private AllDataFilesTableScan(TableOperations ops, Table table, Schema schema, | ||
| TableScanContext context) { | ||
| super(ops, table, schema, MetadataTableType.ALL_FILES, context); | ||
| } | ||
|
|
||
| @Override | ||
| protected TableScan newRefinedScan(TableOperations ops, Table table, Schema schema, TableScanContext context) { | ||
| return new AllDataFilesTableScan(ops, table, schema, context); | ||
| } | ||
|
|
||
| @Override | ||
| protected CloseableIterable<ManifestFile> manifests() { | ||
| return reachableManifests(Snapshot::allManifests); | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,6 +25,7 @@ | |
| import org.apache.iceberg.expressions.Expressions; | ||
| import org.apache.iceberg.io.CloseableIterable; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Iterables; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Sets; | ||
| import org.apache.iceberg.types.Conversions; | ||
| import org.apache.iceberg.types.Types; | ||
| import org.junit.Assert; | ||
|
|
@@ -51,7 +52,10 @@ public static Object[][] parameters() { | |
| { MetadataTableType.FILES, 1 }, | ||
| { MetadataTableType.FILES, 2 }, | ||
| { MetadataTableType.ALL_DATA_FILES, 1 }, | ||
| { MetadataTableType.ALL_DATA_FILES, 2 } | ||
| { MetadataTableType.ALL_DATA_FILES, 2 }, | ||
| { MetadataTableType.ALL_DELETE_FILES, 2 }, | ||
| { MetadataTableType.ALL_FILES, 1 }, | ||
| { MetadataTableType.ALL_FILES, 2 } | ||
| }; | ||
| } | ||
|
|
||
|
|
@@ -95,7 +99,7 @@ public void setupTable() throws Exception { | |
| .commit(); | ||
| } | ||
|
|
||
| if (type.equals(MetadataTableType.ALL_DATA_FILES)) { | ||
| if (allFileTableTest(type)) { | ||
| // Clear all files from current snapshot to test whether 'all' Files tables scans previous files | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Moves file entries to DELETED state | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Removes all entries | ||
|
|
@@ -114,6 +118,10 @@ private Table createMetadataTable() { | |
| return new DeleteFilesTable(table.ops(), table); | ||
| case ALL_DATA_FILES: | ||
| return new AllDataFilesTable(table.ops(), table); | ||
| case ALL_DELETE_FILES: | ||
| return new AllDeleteFilesTable(table.ops(), table); | ||
| case ALL_FILES: | ||
| return new AllFilesTable(table.ops(), table); | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported metadata table type:" + type); | ||
| } | ||
|
|
@@ -129,14 +137,28 @@ private int expectedScanTaskCount(int partitions) { | |
| } | ||
| case DATA_FILES: | ||
| case DELETE_FILES: | ||
| case ALL_DELETE_FILES: | ||
| return partitions; | ||
| case ALL_DATA_FILES: | ||
| return partitions * 2; // ScanTask for Data Manifest in DELETED and ADDED states | ||
| case ALL_FILES: | ||
| if (formatVersion == 1) { | ||
| return partitions * 2; // ScanTask for Data Manifest in DELETED and ADDED states | ||
| } else { | ||
| return partitions * 4; // ScanTask for Delete and Data File in DELETED and ADDED states | ||
| } | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported metadata table type:" + type); | ||
| } | ||
| } | ||
|
|
||
| private boolean allFileTableTest(MetadataTableType tableType) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nit: This might read better as Maybe Given there will be an
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What about |
||
| return Sets.newHashSet(MetadataTableType.ALL_DATA_FILES, | ||
| MetadataTableType.ALL_DATA_FILES, | ||
| MetadataTableType.ALL_FILES) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nit: Consider making this a
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1 for a constant |
||
| .contains(tableType); | ||
| } | ||
|
|
||
| @Test | ||
| public void testNoFilter() { | ||
| Table metadataTable = createMetadataTable(); | ||
|
|
@@ -288,7 +310,7 @@ public void testPartitionSpecEvolutionRemovalV1() { | |
| table.newFastAppend().appendFile(data10).commit(); | ||
| table.newFastAppend().appendFile(data11).commit(); | ||
|
|
||
| if (type.equals(MetadataTableType.ALL_DATA_FILES)) { | ||
| if (allFileTableTest(type)) { | ||
| // Clear all files from current snapshot to test whether 'all' Files tables scans previous files | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Moves file entries to DELETED state | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Removes all entries | ||
|
|
@@ -363,7 +385,7 @@ public void testPartitionSpecEvolutionRemovalV2() { | |
| table.newRowDelta().addDeletes(delete11).commit(); | ||
| } | ||
|
|
||
| if (type.equals(MetadataTableType.ALL_DATA_FILES)) { | ||
| if (allFileTableTest(type)) { | ||
| // Clear all files from current snapshot to test whether 'all' Files tables scans previous files | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Moves file entries to DELETED state | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Removes all entries | ||
|
|
@@ -424,7 +446,7 @@ public void testPartitionSpecEvolutionAdditiveV1() { | |
| table.newFastAppend().appendFile(data10).commit(); | ||
| table.newFastAppend().appendFile(data11).commit(); | ||
|
|
||
| if (type.equals(MetadataTableType.ALL_DATA_FILES)) { | ||
| if (allFileTableTest(type)) { | ||
| // Clear all files from current snapshot to test whether 'all' Files tables scans previous files | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Moves file entries to DELETED state | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Removes all entries | ||
|
|
@@ -499,7 +521,7 @@ public void testPartitionSpecEvolutionAdditiveV2() { | |
| table.newRowDelta().addDeletes(delete11).commit(); | ||
| } | ||
|
|
||
| if (type.equals(MetadataTableType.ALL_DATA_FILES)) { | ||
| if (allFileTableTest(type)) { | ||
| // Clear all files from current snapshot to test whether 'all' Files tables scans previous files | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Moves file entries to DELETED state | ||
| table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit(); // Removes all entries | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: redundant
abeforetable's valid delete files?Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hm, I'm not sure about if its redundant ( 'table' needs an article, right?). How about:
A table implementation that exposes its valid delete files as rows