You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2022/05/11 02:19:23 UTC
[GitHub] [iceberg] stevenzwu opened a new pull request, #4744: Incremental append scan impl
stevenzwu opened a new pull request, #4744:
URL: https://github.com/apache/iceberg/pull/4744
this is a follow-up to PR #4580. will need to rebase this branch when 4580 is merged
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872585816
##########
core/src/test/java/org/apache/iceberg/TestBaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.relocated.com.google.common.collect.Iterables;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestBaseIncrementalAppendScan extends ScanTestBase<IncrementalAppendScan> {
+ public TestBaseIncrementalAppendScan(int formatVersion) {
+ super(formatVersion);
+ }
+
+ @Override
+ protected IncrementalAppendScan newScan() {
+ return table.newIncrementalAppendScan();
+ }
+
+ @Test
+ public void testFromSnapshotInclusive() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotInclusive(snapshotA.snapshotId());
+ Assert.assertEquals(3, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testFromSnapshotExclusiveAndToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotExclusive(snapshotA.snapshotId())
+ .toSnapshot(snapshotB.snapshotId());
+ Assert.assertEquals(1, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .toSnapshot(snapshotB.snapshotId());
+ Assert.assertEquals(2, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testMultipleRootSnapshots() throws Exception {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+
+ table.manageSnapshots()
+ .createBranch("branch", snapshotA.snapshotId())
+ .commit();
+
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ // sleep a milli second so that snapshotD will certainly have a different timestamp
+ Thread.sleep(1);
Review Comment:
We typically use `waitUntilAfter(snapshotC.timestampMillis())` to avoid `Thread.sleep`. It does basically the same thing, but ensures that you have a different timestamp.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hameizi commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
hameizi commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1128384043
@stevenzwu This PR is for "append" snapshot, so for V2 table format we will add like "BaseIncrementalOverwriteScan"?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1128408329
@hameizi there could be a new `IncrementalChangelogScan` interface and an implementation for the more complex CDC like read.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878374352
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ long fromSnapshotId = resolveFromSnapshotIdForListenerEvent(
+ table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotId,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema()));
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ /**
+ * Resolve nullable fromSnapshotIdExclusive to a primitive long type,
+ * as {@link IncrementalScanEvent} only accepts primitive long type.
+ */
+ private static long resolveFromSnapshotIdForListenerEvent(
+ Table table, Long fromSnapshotIdExclusive, long toSnapshotIdInclusive) {
+ if (fromSnapshotIdExclusive != null) {
+ return fromSnapshotIdExclusive;
+ } else {
+ // Resolve to the oldest ancestor of toSnapshotIdInclusive.
+ // This doesn't accurately capture the exclusive behavior of fromSnapshotId.
+ // Alternative is a special value (like 0L) to indicate null fromSnapshotIdExclusive.
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table::snapshot);
+ return oldestAncestorSnapshot != null ? oldestAncestorSnapshot.snapshotId() : toSnapshotIdInclusive;
Review Comment:
Should we check that `oldestAncestorSnapshot.parentId() == null`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r884016115
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
Review Comment:
I like the builder slightly better. but as you said, this is minor. We can leave as-is for now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883890566
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
Review Comment:
good point. added a unit test to cover this scenario
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872932558
##########
api/src/main/java/org/apache/iceberg/Scan.java:
##########
@@ -93,6 +100,18 @@
*/
T planWith(ExecutorService executorService);
+ /**
+ * Returns this scan's projection {@link Schema}.
+ * <p>
+ * If the projection schema was set directly using {@link #project(Schema)}, returns that schema.
+ * <p>
+ * If the projection schema was set by calling {@link #select(Collection)}, returns a projection
+ * schema that includes the selected data fields and any fields used in the filter expression.
+ *
+ * @return this scan's projection schema
+ */
+ Schema schema();
Review Comment:
This is moved from `TableScan` to base `Scan` interface as it is used by `ScanTestBase`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878386157
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
Review Comment:
Probably need to add an inclusive/exclusive flag to context so that we keep track of the requested starting snapshot, D.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883958058
##########
core/src/main/java/org/apache/iceberg/TableScanContext.java:
##########
@@ -168,7 +172,17 @@ Long fromSnapshotId() {
TableScanContext fromSnapshotId(long id) {
return new TableScanContext(snapshotId, rowFilter, ignoreResiduals,
caseSensitive, colStats, projectedSchema, selectedColumns, options, id, toSnapshotId,
- planExecutor);
+ planExecutor, fromSnapshotInclusive);
+ }
+
+ TableScanContext fromSnapshotIdInclusive(long id) {
Review Comment:
agree. since this is a non-public class, it is safe to rename the old setter of `fromSnapshotId` to `fromSnapshotIdExclusive`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883996112
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), false));
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table()::snapshot);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), oldestAncestorSnapshot.snapshotId(),
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), true));
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return appendFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> appendFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
Review Comment:
These have been already defines in `BaseTableScan` for most common methods as we have like 10+ scans so I was just thinking about moving them to `BaseScan`. Up to you, though. I don't feel strongly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883993810
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Review Comment:
As a source, which schema do we expect Iceberg to use? What if I scan for appends and some earlier snapshots doesn't have a column? Or what if earlier snapshots have an extra column? It is a tricky use case but I just want to understand our take on this and what we expect from Iceberg and what we expect from the user.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883755327
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return planFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> planFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
+ }
+
+ private static void notifyIncrementalScanEvent(
+ Table table, TableScanContext context, long toSnapshotIdInclusive) {
+ long fromSnapshotId;
+ boolean fromSnapshotInclusive;
+ if (context.fromSnapshotId() != null) {
+ fromSnapshotId = context.fromSnapshotId();
+ fromSnapshotInclusive = context.fromSnapshotInclusive();
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table::snapshot);
+ fromSnapshotId = oldestAncestorSnapshot.snapshotId();
+ fromSnapshotInclusive = true;
+ }
+
+ Listeners.notifyAll(new IncrementalScanEvent(table.name(), fromSnapshotId,
+ toSnapshotIdInclusive, context.rowFilter(), table.schema(), fromSnapshotInclusive));
+ }
+
+ private static List<Snapshot> appendsBetween(Table table, Long fromSnapshotIdExclusive, long toSnapshotIdInclusive) {
+ List<Snapshot> snapshots = Lists.newArrayList();
+ for (Snapshot snapshot : SnapshotUtil.ancestorsBetween(
Review Comment:
Comment: this does not validation because `planFiles` validated the from snapshot and to snapshot range.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883967787
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
Review Comment:
@aokolnychyi agree with your comment that boolean flag is not clear without looking into the code. Just static methods like `fromSnapshotInclusive` and `fromSnapshotExclusive` is also a little odd to me. Do you think we should provide a builder class instead? Then in the builder class, we can provide methods like `fromSnapshotInclusive` and `fromSnapshotExclusive` methods.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883896076
##########
core/src/main/java/org/apache/iceberg/BaseScan.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.PropertyUtil;
+
+abstract class BaseScan<T extends Scan<T>> implements Scan<T> {
+ private final TableOperations ops;
+ private final Table table;
+ private final Schema schema;
+ private final TableScanContext context;
+
+ protected BaseScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ this.ops = ops;
+ this.table = table;
+ this.schema = schema;
+ this.context = context;
+ }
+
+ protected TableOperations tableOps() {
+ return ops;
+ }
+
+ protected Table table() {
+ return table;
+ }
+
+ protected Schema tableSchema() {
+ return schema;
+ }
+
+ protected TableScanContext context() {
+ return context;
+ }
+
+ protected abstract T newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext);
+
+ @Override
+ public T option(String property, String value) {
+ return newRefinedScan(ops, table, schema, context.withOption(property, value));
+ }
+
+ @Override
+ public T project(Schema projectedSchema) {
+ return newRefinedScan(ops, table, schema, context.project(projectedSchema));
+ }
+
+ @Override
+ public T caseSensitive(boolean caseSensitive) {
+ return newRefinedScan(ops, table, schema, context.setCaseSensitive(caseSensitive));
+ }
+
+ @Override
+ public T includeColumnStats() {
+ return newRefinedScan(ops, table, schema, context.shouldReturnColumnStats(true));
+ }
+
+ @Override
+ public T select(Collection<String> columns) {
+ return newRefinedScan(ops, table, schema, context.selectColumns(columns));
+ }
+
+ @Override
+ public T filter(Expression expr) {
+ return newRefinedScan(ops, table, schema, context.filterRows(Expressions.and(context.rowFilter(), expr)));
+ }
+
+ @Override
+ public T ignoreResiduals() {
+ return newRefinedScan(ops, table, schema, context.ignoreResiduals(true));
+ }
+
+ @Override
+ public T planWith(ExecutorService executorService) {
+ return newRefinedScan(ops, table, schema, context.planWith(executorService));
+ }
+
+ @Override
+ public Schema schema() {
+ return lazyColumnProjection(context, schema);
+ }
+
+ @Override
+ public long targetSplitSize() {
Review Comment:
Not directly related to this PR but can we remove the override of this method in `DataTableScan`?
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Review Comment:
How do we handle schema evolution in incremental scans? Do we always pick the current schema? Do we test it anywhere?
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
Review Comment:
nit: arg formatting
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
Review Comment:
Looks like this block accesses the same variables from `context` quite a few times. Would it make sense to define temp vars? It may make it a little bit easier to read as there will be fewer method calls.
##########
core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java:
##########
@@ -107,6 +120,26 @@ public static Snapshot oldestAncestor(Table table) {
return lastSnapshot;
}
+ /**
+ * Traverses the history and finds the oldest ancestor of the specified snapshot.
+ * <p>
+ * Oldest ancestor is defined as the ancestor snapshot whose parent is null or has been expired.
+ * If the specified snapshot has no parent or parent has been expired,
+ * the specified snapshot itself is returned.
+ *
+ * @param snapshotId the ID of the snapshot to find the oldest ancestor
+ * @param lookup lookup function from snapshot ID to snapshot
+ * @return null if there is no current snapshot in the table, else the oldest Snapshot.
+ */
+ public static Snapshot oldestAncestor(long snapshotId, Function<Long, Snapshot> lookup) {
Review Comment:
Question: shall we call it `oldestAncestorOf`? I guess other methods in this class use this pattern when a snapshot ID is provided explicitly.
##########
core/src/main/java/org/apache/iceberg/BaseScan.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.PropertyUtil;
+
+abstract class BaseScan<T extends Scan<T>> implements Scan<T> {
+ private final TableOperations ops;
+ private final Table table;
+ private final Schema schema;
+ private final TableScanContext context;
+
+ protected BaseScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ this.ops = ops;
+ this.table = table;
+ this.schema = schema;
+ this.context = context;
+ }
+
+ protected TableOperations tableOps() {
+ return ops;
+ }
+
+ protected Table table() {
+ return table;
+ }
+
+ protected Schema tableSchema() {
+ return schema;
+ }
+
+ protected TableScanContext context() {
+ return context;
+ }
+
+ protected abstract T newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext);
+
+ @Override
+ public T option(String property, String value) {
+ return newRefinedScan(ops, table, schema, context.withOption(property, value));
+ }
+
+ @Override
+ public T project(Schema projectedSchema) {
+ return newRefinedScan(ops, table, schema, context.project(projectedSchema));
+ }
+
+ @Override
+ public T caseSensitive(boolean caseSensitive) {
+ return newRefinedScan(ops, table, schema, context.setCaseSensitive(caseSensitive));
+ }
+
+ @Override
+ public T includeColumnStats() {
+ return newRefinedScan(ops, table, schema, context.shouldReturnColumnStats(true));
+ }
+
+ @Override
+ public T select(Collection<String> columns) {
+ return newRefinedScan(ops, table, schema, context.selectColumns(columns));
+ }
+
+ @Override
+ public T filter(Expression expr) {
+ return newRefinedScan(ops, table, schema, context.filterRows(Expressions.and(context.rowFilter(), expr)));
+ }
+
+ @Override
+ public T ignoreResiduals() {
+ return newRefinedScan(ops, table, schema, context.ignoreResiduals(true));
+ }
+
+ @Override
+ public T planWith(ExecutorService executorService) {
+ return newRefinedScan(ops, table, schema, context.planWith(executorService));
+ }
+
+ @Override
+ public Schema schema() {
+ return lazyColumnProjection(context, schema);
+ }
+
+ @Override
+ public long targetSplitSize() {
+ long tableValue = ops.current().propertyAsLong(
+ TableProperties.SPLIT_SIZE,
+ TableProperties.SPLIT_SIZE_DEFAULT);
+ return PropertyUtil.propertyAsLong(context.options(), TableProperties.SPLIT_SIZE, tableValue);
+ }
+
+ @Override
+ public int splitLookback() {
+ int tableValue = ops.current().propertyAsInt(
+ TableProperties.SPLIT_LOOKBACK,
+ TableProperties.SPLIT_LOOKBACK_DEFAULT);
+ return PropertyUtil.propertyAsInt(context.options(), TableProperties.SPLIT_LOOKBACK, tableValue);
+ }
+
+ @Override
+ public long splitOpenFileCost() {
+ long tableValue = ops.current().propertyAsLong(
+ TableProperties.SPLIT_OPEN_FILE_COST,
+ TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT);
+ return PropertyUtil.propertyAsLong(context.options(), TableProperties.SPLIT_OPEN_FILE_COST, tableValue);
+ }
+
+ /**
+ * Resolve the schema to be projected lazily.
+ *
+ * if there are selected columns from scan context, selected columns are projected to the table schema.
Review Comment:
nit: `if` -> `If`
##########
core/src/main/java/org/apache/iceberg/TableScanContext.java:
##########
@@ -168,7 +172,17 @@ Long fromSnapshotId() {
TableScanContext fromSnapshotId(long id) {
return new TableScanContext(snapshotId, rowFilter, ignoreResiduals,
caseSensitive, colStats, projectedSchema, selectedColumns, options, id, toSnapshotId,
- planExecutor);
+ planExecutor, fromSnapshotInclusive);
+ }
+
+ TableScanContext fromSnapshotIdInclusive(long id) {
Review Comment:
Shall we rename the existing methods (`fromSnapshotId`) to indicate they configure the start snapshot exclusive?
We may also set `fromSnapshotInclusive` explicitly in existing methods.
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
Review Comment:
It is not very obvious to have a return statement for the whole method nested in a block that initializes the end snapshot.
Have you considered a separate method for computing the end snapshot, a call to `Preconditions` and a separate if statement for an early return?
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
Review Comment:
nit: The purpose of the true/false flags is not clear without looking into the code. Would it make sense to offer static methods like `fromSnapshotInclusive` and `fromSnapshotExclusive` in `IncrementalScanEvent`?
##########
core/src/main/java/org/apache/iceberg/BaseScan.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.PropertyUtil;
+
+abstract class BaseScan<T extends Scan<T>> implements Scan<T> {
+ private final TableOperations ops;
+ private final Table table;
+ private final Schema schema;
+ private final TableScanContext context;
+
+ protected BaseScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ this.ops = ops;
+ this.table = table;
+ this.schema = schema;
+ this.context = context;
+ }
+
+ protected TableOperations tableOps() {
+ return ops;
+ }
+
+ protected Table table() {
+ return table;
+ }
+
+ protected Schema tableSchema() {
+ return schema;
+ }
+
+ protected TableScanContext context() {
+ return context;
+ }
+
+ protected abstract T newRefinedScan(
Review Comment:
nit: arg formatting
It is not super consistent in the code but most places would format args like this:
<img width="737" alt="image" src="https://user-images.githubusercontent.com/6235869/170771523-96c9060d-19e5-4927-be56-f505c291de32.png">
##########
core/src/main/java/org/apache/iceberg/BaseTableScan.java:
##########
@@ -45,70 +38,46 @@
/**
* Base class for {@link TableScan} implementations.
*/
-abstract class BaseTableScan implements TableScan {
+abstract class BaseTableScan extends BaseScan<TableScan> implements TableScan {
private static final Logger LOG = LoggerFactory.getLogger(BaseTableScan.class);
- private final TableOperations ops;
- private final Table table;
- private final Schema schema;
- private final TableScanContext context;
-
protected BaseTableScan(TableOperations ops, Table table, Schema schema) {
this(ops, table, schema, new TableScanContext());
}
protected BaseTableScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
- this.ops = ops;
- this.table = table;
- this.schema = schema;
- this.context = context;
- }
-
- protected TableOperations tableOps() {
- return ops;
- }
-
- protected Schema tableSchema() {
- return schema;
+ super(ops, table, schema, context);
}
protected Long snapshotId() {
- return context.snapshotId();
+ return context().snapshotId();
}
protected boolean colStats() {
- return context.returnColumnStats();
+ return context().returnColumnStats();
}
protected boolean shouldIgnoreResiduals() {
- return context.ignoreResiduals();
+ return context().ignoreResiduals();
}
protected Collection<String> selectedColumns() {
- return context.selectedColumns();
+ return context().selectedColumns();
}
protected ExecutorService planExecutor() {
- return context.planExecutor();
+ return context().planExecutor();
}
protected Map<String, String> options() {
- return context.options();
- }
-
- protected TableScanContext context() {
- return context;
+ return context().options();
}
- @SuppressWarnings("checkstyle:HiddenField")
- protected abstract TableScan newRefinedScan(TableOperations ops, Table table, Schema schema,
- TableScanContext context);
-
protected abstract CloseableIterable<FileScanTask> doPlanFiles();
@Override
public Table table() {
Review Comment:
Do we still need this implementation here? Can't we consume from the parent directly?
##########
core/src/main/java/org/apache/iceberg/BaseTableScan.java:
##########
@@ -45,70 +38,46 @@
/**
* Base class for {@link TableScan} implementations.
*/
-abstract class BaseTableScan implements TableScan {
+abstract class BaseTableScan extends BaseScan<TableScan> implements TableScan {
private static final Logger LOG = LoggerFactory.getLogger(BaseTableScan.class);
- private final TableOperations ops;
- private final Table table;
- private final Schema schema;
- private final TableScanContext context;
-
protected BaseTableScan(TableOperations ops, Table table, Schema schema) {
this(ops, table, schema, new TableScanContext());
}
protected BaseTableScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
- this.ops = ops;
- this.table = table;
- this.schema = schema;
- this.context = context;
- }
-
- protected TableOperations tableOps() {
- return ops;
- }
-
- protected Schema tableSchema() {
- return schema;
+ super(ops, table, schema, context);
}
protected Long snapshotId() {
- return context.snapshotId();
+ return context().snapshotId();
}
protected boolean colStats() {
- return context.returnColumnStats();
+ return context().returnColumnStats();
}
protected boolean shouldIgnoreResiduals() {
- return context.ignoreResiduals();
+ return context().ignoreResiduals();
}
protected Collection<String> selectedColumns() {
Review Comment:
Is this still being used?
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), false));
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table()::snapshot);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), oldestAncestorSnapshot.snapshotId(),
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), true));
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return appendFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> appendFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
Review Comment:
This block also accesses a lot of information on `context()`. There were some helper methods in `BaseTableScan`. What about moving them to `BaseScan` and using here?
```
protected Long snapshotId() {
return context().snapshotId();
}
protected boolean colStats() {
return context().returnColumnStats();
}
protected boolean shouldIgnoreResiduals() {
return context().ignoreResiduals();
}
@Override
public Expression filter() {
return context().rowFilter();
}
@Override
public boolean isCaseSensitive() {
return context().caseSensitive();
}
protected ExecutorService planExecutor() {
return context().planExecutor();
}
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872907901
##########
api/src/test/java/org/apache/iceberg/TestHelpers.java:
##########
@@ -42,6 +42,17 @@ public class TestHelpers {
private TestHelpers() {
}
+ /**
+ * Wait in a tight check loop until system clock is past {@code timestampMillis}
+ */
+ public static long waitUntilAfter(long timestampMillis) {
Review Comment:
This is copied from `SparkTestBase`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883752495
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
Review Comment:
t0 : A - B - C
t1 : B - C
scan( A, C, exclusive )
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1139991489
@aokolnychyi, Sorry, I didn't realize you were reviewing at the same time. I think a follow-up is a good idea, I had a comment as well. I just wanted to get this in to unblock the rest since I think it's correct.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883968444
##########
core/src/main/java/org/apache/iceberg/BaseTableScan.java:
##########
@@ -45,70 +38,46 @@
/**
* Base class for {@link TableScan} implementations.
*/
-abstract class BaseTableScan implements TableScan {
+abstract class BaseTableScan extends BaseScan<TableScan> implements TableScan {
private static final Logger LOG = LoggerFactory.getLogger(BaseTableScan.class);
- private final TableOperations ops;
- private final Table table;
- private final Schema schema;
- private final TableScanContext context;
-
protected BaseTableScan(TableOperations ops, Table table, Schema schema) {
this(ops, table, schema, new TableScanContext());
}
protected BaseTableScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
- this.ops = ops;
- this.table = table;
- this.schema = schema;
- this.context = context;
- }
-
- protected TableOperations tableOps() {
- return ops;
- }
-
- protected Schema tableSchema() {
- return schema;
+ super(ops, table, schema, context);
}
protected Long snapshotId() {
- return context.snapshotId();
+ return context().snapshotId();
}
protected boolean colStats() {
- return context.returnColumnStats();
+ return context().returnColumnStats();
}
protected boolean shouldIgnoreResiduals() {
- return context.ignoreResiduals();
+ return context().ignoreResiduals();
}
protected Collection<String> selectedColumns() {
- return context.selectedColumns();
+ return context().selectedColumns();
}
protected ExecutorService planExecutor() {
- return context.planExecutor();
+ return context().planExecutor();
}
protected Map<String, String> options() {
- return context.options();
- }
-
- protected TableScanContext context() {
- return context;
+ return context().options();
}
- @SuppressWarnings("checkstyle:HiddenField")
- protected abstract TableScan newRefinedScan(TableOperations ops, Table table, Schema schema,
- TableScanContext context);
-
protected abstract CloseableIterable<FileScanTask> doPlanFiles();
@Override
public Table table() {
Review Comment:
yeah. `BaseScan#table()` is a protected method. this is a public method.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878372934
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ long fromSnapshotId = resolveFromSnapshotIdForListenerEvent(
+ table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotId,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema()));
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ /**
+ * Resolve nullable fromSnapshotIdExclusive to a primitive long type,
+ * as {@link IncrementalScanEvent} only accepts primitive long type.
+ */
+ private static long resolveFromSnapshotIdForListenerEvent(
+ Table table, Long fromSnapshotIdExclusive, long toSnapshotIdInclusive) {
+ if (fromSnapshotIdExclusive != null) {
+ return fromSnapshotIdExclusive;
+ } else {
+ // Resolve to the oldest ancestor of toSnapshotIdInclusive.
+ // This doesn't accurately capture the exclusive behavior of fromSnapshotId.
+ // Alternative is a special value (like 0L) to indicate null fromSnapshotIdExclusive.
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table::snapshot);
+ return oldestAncestorSnapshot != null ? oldestAncestorSnapshot.snapshotId() : toSnapshotIdInclusive;
Review Comment:
Can we also add an inclusive/exclusive flag to the event so that we are correctly labelling this case?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878585592
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -53,14 +53,7 @@ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
Preconditions.checkArgument(startingSnapshotInclusive != null,
"Cannot find the starting snapshot: %d", fromSnapshotId);
- // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
- // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
- // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
- if (startingSnapshotInclusive.parentId() != null) {
- return newRefinedScan(tableOps(), table(), schema(),
- context().fromSnapshotId(startingSnapshotInclusive.parentId()));
- }
- return this;
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
Review Comment:
this was a bug that was caught when adding a new test coverage
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872932456
##########
api/src/main/java/org/apache/iceberg/Scan.java:
##########
@@ -84,6 +84,13 @@
*/
T filter(Expression expr);
+ /**
+ * Create a new scan from this that applies data filtering to files but not to rows in those files.
+ *
+ * @return a new scan based on this that does not filter rows in files.
+ */
+ T ignoreResiduals();
Review Comment:
this is missed in the last API PR
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r876134820
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ long fromSnapshotId = resolveFromSnapshotIdForListenerEvent(
+ table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotId,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema()));
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
Review Comment:
This impl works with appends only table. For row-level deletes, we would need another `IncrementalChangelogScan` interface and implementation.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878586330
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -79,26 +72,50 @@ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
@Override
public CloseableIterable<FileScanTask> planFiles() {
- // Return an empty iterable when table is empty
- if (table().currentSnapshot() == null) {
- return CloseableIterable.empty();
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
}
- long toSnapshotIdInclusive = context().toSnapshotId() != null ?
- context().toSnapshotId() : table().currentSnapshot().snapshotId();
+
if (context().fromSnapshotId() != null) {
Preconditions.checkArgument(
SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
- "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
context().fromSnapshotId(), toSnapshotIdInclusive);
}
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
+
// appendsBetween handles null fromSnapshotId (exclusive) properly
- List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
if (snapshots.isEmpty()) {
return CloseableIterable.empty();
}
+ return planFilesFromSnapshots(snapshots);
Review Comment:
this method was extracted to avoid checkstyle error on CyclomaticComplexity
```
[ant:checkstyle] [ERROR] /Users/stevenwu/ws/iceberg/core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:73:3: Cyclomatic Complexity is 13 (max allowed is 12). [CyclomaticComplexity]
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878391857
##########
core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java:
##########
@@ -107,6 +107,19 @@ public static Snapshot oldestAncestor(Table table) {
return lastSnapshot;
}
+ /**
+ * Traverses the history and finds the oldest ancestor of snapshotId.
+ * @return null if there is no current snapshot in the table, else the oldest Snapshot.
+ */
+ public static Snapshot oldestAncestor(long snapshotId, Function<Long, Snapshot> lookup) {
+ Snapshot lastSnapshot = null;
+ for (Snapshot snapshot : ancestorsOf(snapshotId, lookup)) {
Review Comment:
Be aware that this includes the snapshot for `snapshotId`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878391008
##########
core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java:
##########
@@ -107,6 +107,19 @@ public static Snapshot oldestAncestor(Table table) {
return lastSnapshot;
}
+ /**
+ * Traverses the history and finds the oldest ancestor of snapshotId.
+ * @return null if there is no current snapshot in the table, else the oldest Snapshot.
Review Comment:
Also needs a `@param` for `snapshotId` and `lookup`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883956271
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Review Comment:
Good question. I think it is outside the scope of the scan/source.
Let's use the more common streaming source (kafka with Avro record). The problem of schema evolution is outside the scope of Kafka source. If the schema is evolved in a backward compatible, the consumers won't break. If the schema evolution is incompatible, then consumers should be notified and the breaking change need to be coordinated. If there is a Avro schema registry, notification service can be provided by schema registry or some other control plane.
Now coming to Iceberg streaming source, I would assume similar expectation. maybe some control plane service can notify consumers of table schema change. Then consumers can take proper action if needed
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872591274
##########
core/src/test/java/org/apache/iceberg/TestBaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.relocated.com.google.common.collect.Iterables;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestBaseIncrementalAppendScan extends ScanTestBase<IncrementalAppendScan> {
+ public TestBaseIncrementalAppendScan(int formatVersion) {
+ super(formatVersion);
+ }
+
+ @Override
+ protected IncrementalAppendScan newScan() {
+ return table.newIncrementalAppendScan();
+ }
+
+ @Test
+ public void testFromSnapshotInclusive() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotInclusive(snapshotA.snapshotId());
+ Assert.assertEquals(3, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testFromSnapshotExclusiveAndToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotExclusive(snapshotA.snapshotId())
+ .toSnapshot(snapshotB.snapshotId());
+ Assert.assertEquals(1, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshotB = table.currentSnapshot();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot snapshotC = table.currentSnapshot();
+
+ IncrementalAppendScan scan = newScan()
+ .toSnapshot(snapshotB.snapshotId());
+ Assert.assertEquals(2, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testMultipleRootSnapshots() throws Exception {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot snapshotA = table.currentSnapshot();
+
+ table.manageSnapshots()
+ .createBranch("branch", snapshotA.snapshotId())
+ .commit();
Review Comment:
Remove the branch. Add:
```java
long expireA = waitUntilAfter(snapshotA.timestampMillis());
```
Then use `table.newFastAppend().appendFile(FILE_D).stageOnly().commit()` to create the unrelated commit for D. And then create B and C.
And expire A with:
```java
table.expireSnapshots().expireOlderThan(expireA).commit();
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hameizi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
hameizi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r874402526
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ long fromSnapshotId = resolveFromSnapshotIdForListenerEvent(
+ table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotId,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema()));
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
Review Comment:
`manifestGroup.planFiles()` will apply all delete-file whose seq number greater than datafile even though the delete file is not belong to the snapshots we scan, is it the intended effect?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878398311
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
Review Comment:
I think %s would fill in the values.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883740841
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
Review Comment:
I'm not sure this is the right behavior. If you try to scan an empty table, then it should return 0 tasks instead of failing.
I think the case where this should fail is when the from snapshot ID is set and there is no to snapshot.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883743794
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
Review Comment:
This doesn't handle the exclusive case by checking parent ID, so I think you need to loop over the snapshots and check parent ID.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883981634
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -77,35 +76,52 @@ public CloseableIterable<FileScanTask> planFiles() {
toSnapshotIdInclusive = context().toSnapshotId();
} else if (table().currentSnapshot() != null) {
toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
} else {
- throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
}
-
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
Review Comment:
addressed in PR #4886
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883995287
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
Review Comment:
I mentioned separate methods as we usually prefer a named method to give some meaning to a boolean flag but I agree there are quite some args in this case so such it is questionable. Builder sounds good but this point is minor. Up to you. I think we can leave as-is and don't worry about it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1139986187
Thanks, @stevenzwu!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872580847
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,247 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+import org.apache.iceberg.events.IncrementalAppendScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan implements IncrementalAppendScan {
+
+ private final TableOperations ops;
+ private final Table table;
+ private TableScanContext context;
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, TableScanContext context) {
+ this.ops = ops;
+ this.table = table;
+ this.context = context;
+ }
+
+ @Override
+ public IncrementalAppendScan option(String property, String value) {
+ this.context = context.withOption(property, value);
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan project(Schema projectedSchema) {
+ this.context = context.project(projectedSchema);
+ return this;
Review Comment:
The `TableScan` interface is immutable and we should follow the same pattern here by returning a new scan object with the new configuration merged in. The `context` already does this (which is why you have to assign the result to `context`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r881229437
##########
api/src/main/java/org/apache/iceberg/events/IncrementalScanEvent.java:
##########
@@ -60,4 +62,8 @@ public Expression filter() {
public Schema projection() {
return projection;
}
+
+ public boolean fromSnapshotInclusive() {
Review Comment:
agree. updated
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883965470
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
Review Comment:
sounds good. will make the change
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r884027693
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Review Comment:
In Kafka source case, users can pin the read Avro schema. As long as schema evolution is backward compatible (like an extra column), this is fine. If users don't pin read schema, only Avro GenericRecord is possible with a schema registry. Kafka record contains a schema id or fingerprint that consumers can use to look up schema from registry. SpecificRecord implicitly pinned the read schema to the schema used to generate the SpecificRecord code.
Similarly with Flink Iceberg source (as an example), users can set read/projected schema. Then the Flink reader should emit `RowData` with fields only from the pinned read schema. if users don't pin the read schema, current table schema is used as read schema. Read schema stays the same until the next time the Flink job is redeployed and then read schema is refreshed to the table schema again.
If a mandatory field is deleted (incompatible schema change), then consumer apps may be broken if the deleted column is accessed.
Here is probably a more tricky scenario that would fail
- users don't pin read schema
- a new required column is added to Iceberg table schema
- read scanned some old snapshots/data where is new required column is not present
Either users should pin read schema or users need to make sure not to scan old data where current table is incompatible. I would think adding a required column is also an incompatible schema evolution as it is incompatible to older data.
cc @rdblue @kbendick @Fokko for more inputs
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1137482818
I'd be interested to take a look today as well.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883753570
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
Review Comment:
Add `Long fromSnapshotId = context().fromSnapshotId()`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883972758
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), false));
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table()::snapshot);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), oldestAncestorSnapshot.snapshotId(),
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), true));
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return appendFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> appendFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
Review Comment:
I am not sure adding a bunch help methods like these makes the code simpler
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#issuecomment-1139989014
@rdblue @stevenzwu, I had some minor comments about styling in a few places. They don't affect correctness but it would be great to fix some of those in a follow-up.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883745570
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
Review Comment:
I think this should either pass the information for the event, or pass the context, but not both. It's unclear where information should come from and we've already done the analysis to know the starting snapshot here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883744468
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
Review Comment:
This calls `fromSnapshotId` in a lot of places. Can you store that in a variable?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883746383
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return planFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> planFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
+ }
+
+ private static void notifyIncrementalScanEvent(
+ Table table, TableScanContext context, long toSnapshotIdInclusive) {
+ long fromSnapshotId;
+ boolean fromSnapshotInclusive;
+ if (context.fromSnapshotId() != null) {
+ fromSnapshotId = context.fromSnapshotId();
+ fromSnapshotInclusive = context.fromSnapshotInclusive();
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table::snapshot);
Review Comment:
Looks like this is redoing work that has already been done in `planFiles`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883756109
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ }
+
+
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return planFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> planFilesFromSnapshots(List<Snapshot> snapshots) {
Review Comment:
Can we have a more specific name, like `appendFilesFromSnapshots`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878381671
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
Review Comment:
With the `currentSnapshot()` check removed above, you'll need to update this:
```java
long toSnapshotIdInclusive;
if (context.toSnapshotId() != null) {
toSnapshotIdInclusive = context.toSnapshotId();
} else if (table.currentSnapshot() != null) {
toSnapshotIdInclusive = table.currentSnapshot().snapshotId();
} else {
// throw exception
}
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878374352
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
+ .filterData(context().rowFilter())
+ .filterManifestEntries(
+ manifestEntry ->
+ snapshotIds.contains(manifestEntry.snapshotId()) &&
+ manifestEntry.status() == ManifestEntry.Status.ADDED)
+ .specsById(tableOps().current().specsById())
+ .ignoreDeleted();
+
+ if (context().ignoreResiduals()) {
+ manifestGroup = manifestGroup.ignoreResiduals();
+ }
+
+ long fromSnapshotId = resolveFromSnapshotIdForListenerEvent(
+ table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotId,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema()));
+
+ if (manifests.size() > 1 &&
+ (DataTableScan.PLAN_SCANS_WITH_WORKER_POOL || context().planWithCustomizedExecutor())) {
+ manifestGroup = manifestGroup.planWith(context().planExecutor());
+ }
+
+ return manifestGroup.planFiles();
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ /**
+ * Resolve nullable fromSnapshotIdExclusive to a primitive long type,
+ * as {@link IncrementalScanEvent} only accepts primitive long type.
+ */
+ private static long resolveFromSnapshotIdForListenerEvent(
+ Table table, Long fromSnapshotIdExclusive, long toSnapshotIdInclusive) {
+ if (fromSnapshotIdExclusive != null) {
+ return fromSnapshotIdExclusive;
+ } else {
+ // Resolve to the oldest ancestor of toSnapshotIdInclusive.
+ // This doesn't accurately capture the exclusive behavior of fromSnapshotId.
+ // Alternative is a special value (like 0L) to indicate null fromSnapshotIdExclusive.
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table::snapshot);
+ return oldestAncestorSnapshot != null ? oldestAncestorSnapshot.snapshotId() : toSnapshotIdInclusive;
Review Comment:
Should we check that `oldestAncestorSnapshot.parentId() == fromSnapshotIdExclusive`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878395665
##########
core/src/test/java/org/apache/iceberg/TestBaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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 java.util.List;
+import java.util.stream.Collectors;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestBaseIncrementalAppendScan extends ScanTestBase<IncrementalAppendScan> {
+ public TestBaseIncrementalAppendScan(int formatVersion) {
+ super(formatVersion);
+ }
+
+ @Override
+ protected IncrementalAppendScan newScan() {
+ return table.newIncrementalAppendScan();
+ }
+
+ @Test
+ public void testFromSnapshotInclusive() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotAId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ long snapshotBId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ long snapshotCId = table.currentSnapshot().snapshotId();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotInclusive(snapshotAId);
+ Assert.assertEquals(3, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testFromSnapshotExclusiveAndToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotAId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ long snapshotBId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ long snapshotCId = table.currentSnapshot().snapshotId();
+
+ IncrementalAppendScan scan = newScan()
+ .fromSnapshotExclusive(snapshotAId)
+ .toSnapshot(snapshotBId);
+ Assert.assertEquals(1, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testToSnapshot() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotAId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ long snapshotBId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_C).commit();
+ long snapshotCId = table.currentSnapshot().snapshotId();
+
+ IncrementalAppendScan scan = newScan()
+ .toSnapshot(snapshotBId);
+ Assert.assertEquals(2, Iterables.size(scan.planFiles()));
+ }
+
+ @Test
+ public void testMultipleRootSnapshots() throws Exception {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotAId = table.currentSnapshot().snapshotId();
+ long expireTimestampSnapshotA = TestHelpers.waitUntilAfter(table.currentSnapshot().timestampMillis());
+
+ // append file B in a staged branch
+ table.newFastAppend().appendFile(FILE_B).stageOnly().commit();
Review Comment:
You can also call the operation's `apply()` method to get the snapshot ID early.
```java
AppendFiles append = table.newFastAppend()...;
Snapshot snapshotWithB = append.apply();
append.commit();
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878586330
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -79,26 +72,50 @@ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
@Override
public CloseableIterable<FileScanTask> planFiles() {
- // Return an empty iterable when table is empty
- if (table().currentSnapshot() == null) {
- return CloseableIterable.empty();
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else {
+ throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
}
- long toSnapshotIdInclusive = context().toSnapshotId() != null ?
- context().toSnapshotId() : table().currentSnapshot().snapshotId();
+
if (context().fromSnapshotId() != null) {
Preconditions.checkArgument(
SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
- "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ "Starting snapshot (%s) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotInclusive() ? "inclusive" : "exclusive",
context().fromSnapshotId(), toSnapshotIdInclusive);
}
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null && context().fromSnapshotInclusive()) {
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ // appendsBetween handles null fromSnapshotIdExclusive properly by finding the ancestors of end snapshot.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ }
+
+ notifyIncrementalScanEvent(table(), context(), toSnapshotIdInclusive);
+
// appendsBetween handles null fromSnapshotId (exclusive) properly
- List<Snapshot> snapshots = appendsBetween(table(), context().fromSnapshotId(), toSnapshotIdInclusive);
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
if (snapshots.isEmpty()) {
return CloseableIterable.empty();
}
+ return planFilesFromSnapshots(snapshots);
Review Comment:
this method was extracted to avoid checkstyle error
```
[ant:checkstyle] [ERROR] /Users/stevenwu/ws/iceberg/core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:73:3: Cyclomatic Complexity is 13 (max allowed is 12). [CyclomaticComplexity]
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872908160
##########
core/src/test/java/org/apache/iceberg/ScanTestBase.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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 java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+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.types.Types;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public abstract class ScanTestBase<T extends Scan<T>> extends TableTestBase {
Review Comment:
This is refactored out of `TestDataTableScan`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r872932403
##########
api/src/main/java/org/apache/iceberg/Scan.java:
##########
@@ -114,4 +133,19 @@
* @return an Iterable of tasks for this scan
*/
CloseableIterable<CombinedScanTask> planTasks();
+
+ /**
+ * Returns the target split size for this scan.
+ */
+ long targetSplitSize();
Review Comment:
I moved those 3 split methods so that we can share the implementations in `BaseScan` class.
Alternatively, we can move them into a static util class. Would love to get some feedback here.
```
public class ScanUtil {
public static long targetSplitSize(TableOperations ops, TableScanContext context) {
long tableValue = ops.current().propertyAsLong(
TableProperties.SPLIT_SIZE,
TableProperties.SPLIT_SIZE_DEFAULT);
return PropertyUtil.propertyAsLong(context.options(), TableProperties.SPLIT_SIZE, tableValue);
}
}
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r879860526
##########
api/src/main/java/org/apache/iceberg/events/IncrementalScanEvent.java:
##########
@@ -60,4 +62,8 @@ public Expression filter() {
public Schema projection() {
return projection;
}
+
+ public boolean fromSnapshotInclusive() {
Review Comment:
Would it be more clear for this to be `isFromSnapshotInclusive`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878384390
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
Review Comment:
Here's a case I'm thinking of:
```
A - B - C <-- current
D - E <-- branch
```
If D's parent is null and the `toSnapshotId` is C. If the from snapshot is D, then the exclusive one will be null, D's parent. But D is not in C's ancestors.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878382588
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
+ return CloseableIterable.empty();
+ }
+
+ long toSnapshotIdInclusive = context().toSnapshotId() != null ?
+ context().toSnapshotId() : table().currentSnapshot().snapshotId();
+ if (context().fromSnapshotId() != null) {
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %d is not an ancestor of end snapshot %d",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
Review Comment:
If the `fromSnapshotId` is null then we need to validate that the oldest ancestor's parent is null.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878380212
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Snapshot startingSnapshotInclusive = table().snapshot(fromSnapshotId);
+ Preconditions.checkArgument(startingSnapshotInclusive != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ // To achieve inclusive behavior, we pass the parent snapshot id to the TableScanContext.
+ // If parent id is null, leave the TableScanContext#fromSnapshotId as null.
+ // This way, TableScanContext#fromSnapshotId always capture the exclusive behavior.
+ if (startingSnapshotInclusive.parentId() != null) {
+ return newRefinedScan(tableOps(), table(), schema(),
+ context().fromSnapshotId(startingSnapshotInclusive.parentId()));
+ }
+ return this;
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ // Return an empty iterable when table is empty
+ if (table().currentSnapshot() == null) {
Review Comment:
I think we can omit this. It was originally an optimization, but now that the format supports branches, there are cases where the main branch is not set, but the range of snapshots to scan is actually valid.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r878393504
##########
core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java:
##########
@@ -107,6 +107,19 @@ public static Snapshot oldestAncestor(Table table) {
return lastSnapshot;
}
+ /**
+ * Traverses the history and finds the oldest ancestor of snapshotId.
Review Comment:
Note that this may return `snapshotId` if that snapshot is the oldest known snapshot (its parent is null or has been expired).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883942065
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -77,35 +76,52 @@ public CloseableIterable<FileScanTask> planFiles() {
toSnapshotIdInclusive = context().toSnapshotId();
} else if (table().currentSnapshot() != null) {
toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
} else {
- throw new IllegalArgumentException("End snapshot is not set and table has no current snapshot");
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
}
-
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
Review Comment:
Can you refactor out all the calls to `context().fromSnapshotId()`? I think it would be more readable to have a variable.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue merged pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
rdblue merged PR #4744:
URL: https://github.com/apache/iceberg/pull/4744
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r884015804
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(fromSnapshotId) != null,
+ "Cannot find the starting snapshot: %d", fromSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotIdInclusive(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotExclusive(long fromSnapshotId) {
+ // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied.
+ // as fromSnapshotId could be matched to a parent snapshot that is already expired
+ return newRefinedScan(tableOps(), table(), schema(), context().fromSnapshotId(fromSnapshotId));
+ }
+
+ @Override
+ public IncrementalAppendScan toSnapshot(long toSnapshotId) {
+ Preconditions.checkArgument(table().snapshot(toSnapshotId) != null,
+ "Cannot find end snapshot: %d", toSnapshotId);
+ return newRefinedScan(tableOps(), table(), schema(), context().toSnapshotId(toSnapshotId));
+ }
+
+ @Override
+ public CloseableIterable<FileScanTask> planFiles() {
+ long toSnapshotIdInclusive;
+ if (context().toSnapshotId() != null) {
+ toSnapshotIdInclusive = context().toSnapshotId();
+ } else if (table().currentSnapshot() != null) {
+ toSnapshotIdInclusive = table().currentSnapshot().snapshotId();
+ } else if (context().fromSnapshotId() != null) {
+ throw new IllegalArgumentException(
+ "Invalid config: end snapshot is not set, start snapshot is set, and table has no current snapshot");
+ } else {
+ // It is an empty table (no current snapshot). Both from and to snapshots aren't set either.
+ // In this case, listener notification is also skipped
+ return CloseableIterable.empty();
+ }
+
+ // fromSnapshotIdExclusive can be null. appendsBetween handles null fromSnapshotIdExclusive properly
+ // by finding the oldest ancestor of end snapshot.
+ Long fromSnapshotIdExclusive = context().fromSnapshotId();
+ if (context().fromSnapshotId() != null) {
+ if (context().fromSnapshotInclusive()) {
+ // validate the fromSnapshotId is an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (inclusive) %s is not an ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot id, which can be null.
+ fromSnapshotIdExclusive = table().snapshot(context().fromSnapshotId()).parentId();
+ } else {
+ // validate the parent snapshot id an ancestor of toSnapshotId
+ Preconditions.checkArgument(
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, context().fromSnapshotId()),
+ "Starting snapshot (exclusive) %s is not a parent ancestor of end snapshot %s",
+ context().fromSnapshotId(), toSnapshotIdInclusive);
+ }
+ }
+
+ if (fromSnapshotIdExclusive != null) {
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), fromSnapshotIdExclusive,
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), false));
+ } else {
+ Snapshot oldestAncestorSnapshot = SnapshotUtil.oldestAncestor(toSnapshotIdInclusive, table()::snapshot);
+ Listeners.notifyAll(new IncrementalScanEvent(table().name(), oldestAncestorSnapshot.snapshotId(),
+ toSnapshotIdInclusive, context().rowFilter(), table().schema(), true));
+ }
+
+ // appendsBetween handles null fromSnapshotId (exclusive) properly
+ List<Snapshot> snapshots = appendsBetween(table(), fromSnapshotIdExclusive, toSnapshotIdInclusive);
+ if (snapshots.isEmpty()) {
+ return CloseableIterable.empty();
+ }
+
+ return appendFilesFromSnapshots(snapshots);
+ }
+
+ @Override
+ public CloseableIterable<CombinedScanTask> planTasks() {
+ CloseableIterable<FileScanTask> fileScanTasks = planFiles();
+ CloseableIterable<FileScanTask> splitFiles = TableScanUtil.splitFiles(fileScanTasks, targetSplitSize());
+ return TableScanUtil.planTasks(splitFiles, targetSplitSize(), splitLookback(), splitOpenFileCost());
+ }
+
+ private CloseableIterable<FileScanTask> appendFilesFromSnapshots(List<Snapshot> snapshots) {
+ Set<Long> snapshotIds = Sets.newHashSet(Iterables.transform(snapshots, Snapshot::snapshotId));
+ Set<ManifestFile> manifests = FluentIterable
+ .from(snapshots)
+ .transformAndConcat(Snapshot::dataManifests)
+ .filter(manifestFile -> snapshotIds.contains(manifestFile.snapshotId()))
+ .toSet();
+
+ ManifestGroup manifestGroup = new ManifestGroup(tableOps().io(), manifests)
+ .caseSensitive(context().caseSensitive())
+ .select(context().returnColumnStats() ? DataTableScan.SCAN_WITH_STATS_COLUMNS : DataTableScan.SCAN_COLUMNS)
Review Comment:
I missed that `BaseTableScan` already has many helper methods like these. I am ok to move the protected helper method to `BaseScan`. I am not opinionated either way.. A couple of the helper methods (like `filter` and `isCaseSensitive`) from `BaseTableScan` are public and have to stay.
@rdblue do you have any preference?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r883971157
##########
core/src/main/java/org/apache/iceberg/BaseTableScan.java:
##########
@@ -45,70 +38,46 @@
/**
* Base class for {@link TableScan} implementations.
*/
-abstract class BaseTableScan implements TableScan {
+abstract class BaseTableScan extends BaseScan<TableScan> implements TableScan {
private static final Logger LOG = LoggerFactory.getLogger(BaseTableScan.class);
- private final TableOperations ops;
- private final Table table;
- private final Schema schema;
- private final TableScanContext context;
-
protected BaseTableScan(TableOperations ops, Table table, Schema schema) {
this(ops, table, schema, new TableScanContext());
}
protected BaseTableScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
- this.ops = ops;
- this.table = table;
- this.schema = schema;
- this.context = context;
- }
-
- protected TableOperations tableOps() {
- return ops;
- }
-
- protected Schema tableSchema() {
- return schema;
+ super(ops, table, schema, context);
}
protected Long snapshotId() {
- return context.snapshotId();
+ return context().snapshotId();
}
protected boolean colStats() {
- return context.returnColumnStats();
+ return context().returnColumnStats();
}
protected boolean shouldIgnoreResiduals() {
- return context.ignoreResiduals();
+ return context().ignoreResiduals();
}
protected Collection<String> selectedColumns() {
Review Comment:
doesn't looks like it is used based on code search. will remove it in the follow up
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #4744: Core: Incremental append scan impl
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on code in PR #4744:
URL: https://github.com/apache/iceberg/pull/4744#discussion_r886197420
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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 java.util.List;
+import java.util.Set;
+import org.apache.iceberg.events.IncrementalScanEvent;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.apache.iceberg.util.TableScanUtil;
+
+class BaseIncrementalAppendScan extends BaseScan<IncrementalAppendScan> implements IncrementalAppendScan {
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table) {
+ this(ops, table, table.schema(), new TableScanContext());
+ }
+
+ BaseIncrementalAppendScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
+ super(ops, table, schema, context);
+ }
+
+ @Override
+ protected IncrementalAppendScan newRefinedScan(
+ TableOperations newOps, Table newTable, Schema newSchema, TableScanContext newContext) {
+ return new BaseIncrementalAppendScan(newOps, newTable, newSchema, newContext);
+ }
+
+ @Override
+ public IncrementalAppendScan fromSnapshotInclusive(long fromSnapshotId) {
Review Comment:
Thanks for the context. I think the current behavior in the incremental scan matches the expectation.
> I would think adding a required column is also an incompatible schema evolution
That's true. Iceberg doesn't allow adding required columns by default unless the user explicitly acknowledged it would not break older data files.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org