pvary commented on code in PR #15996: URL: https://github.com/apache/iceberg/pull/15996#discussion_r3234921356
########## flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java: ########## @@ -0,0 +1,663 @@ +/* + * 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.flink.maintenance.operator; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateSnapshotContext; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.util.OutputTag; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotChanges; +import org.apache.iceberg.Table; +import org.apache.iceberg.flink.TableLoader; +import org.apache.iceberg.flink.maintenance.api.Trigger; +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.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.ContentFileUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Planner for the equality delete conversion pipeline. For each trigger, it picks the oldest + * staging snapshot that hasn't been converted yet and emits {@link ReadCommand}s describing the + * files its downstream readers and workers must process. + * + * <p>Each trigger runs two steps in order: + * + * <ol> + * <li>{@link #ensureIndexCurrent}: updates {@link #lastStagingSnapshotId} from main's history, + * bootstraps the worker index from main on first run, and reindexes when external commits + * (e.g. compaction) have advanced main past the currently-indexed snapshot. + * <li>{@link #processStagingSnapshot}: resolve the chosen staging snapshot's eq deletes against + * the (now-current) index, pass through any DV files, and index the snapshot's new data files + * for the next cycle. + * </ol> + * + * Watermarks separate phases that gate the worker's keyed state. The contract is documented on + * {@link #advancePhase()}. + * + * <p>An {@link EqualityConvertPlanResult} with the current cycle's metadata is emitted via the + * {@link #METADATA_STREAM} side output after the read commands. + * + * <p>Assumes a single equality-field set supplied via the builder; staging eq-deletes with a + * different {@code equalityFieldIds} fail fast in {@link #retrieveStagingFiles}. Concurrent writes + * on the target branch are handled by {@link #ensureIndexCurrent} reindexing from the new main + * snapshot; commit-time conflicts are caught by {@code RowDelta.validateFromSnapshot}. + */ +@Internal +public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand> + implements OneInputStreamOperator<Trigger, ReadCommand> { + + private static final Logger LOG = LoggerFactory.getLogger(EqualityConvertPlanner.class); + + public static final OutputTag<EqualityConvertPlanResult> METADATA_STREAM = + new OutputTag<>("metadata-stream") {}; + + private static final String PROCESSED_EQ_DELETE_FILE_NUM_METRIC = "processedEqDeleteFileNum"; + private static final String PROCESSED_STAGING_SNAPSHOT_NUM_METRIC = "processedStagingSnapshotNum"; + private static final String SKIPPED_NO_OP_CYCLES_METRIC = "skippedNoOpCycles"; + private static final String REINDEX_COUNT_METRIC = "reindexCount"; + + private final String tableName; + private final String taskName; + private final int taskIndex; + private final TableLoader tableLoader; + private final String stagingBranch; + private final String targetBranch; + // Equality-field-id list the worker keys on. Supplied via the builder; every staging + // eq-delete's equalityFieldIds() must match exactly. + private final List<Integer> eqFieldIds; + + // Main snapshot id the worker's index reflects. + private transient ListState<Long> indexSnapshotState; + + private transient Table table; + + private transient Long lastMainSnapshotId; + private transient Long lastStagingSnapshotId; + private transient Long indexSnapshotId; + + private transient long nextPhaseTs; + + // Per-trigger cache of SnapshotChanges to avoid re-parsing manifest lists. + private transient Map<Long, SnapshotChanges> snapshotChangesCache; + + private transient Counter processedEqDeleteFileNumCounter; + private transient Counter processedStagingSnapshotNumCounter; + private transient Counter skippedNoOpCyclesCounter; + private transient Counter reindexCounter; + + public EqualityConvertPlanner( + String tableName, + String taskName, + int taskIndex, + TableLoader tableLoader, + String stagingBranch, + String targetBranch, + List<Integer> eqFieldIds) { + this.tableName = tableName; + this.taskName = taskName; + this.taskIndex = taskIndex; + this.tableLoader = tableLoader; + this.stagingBranch = stagingBranch; + this.targetBranch = targetBranch; + Preconditions.checkArgument( + eqFieldIds != null && !eqFieldIds.isEmpty(), "eqFieldIds must not be null or empty"); + this.eqFieldIds = Collections.unmodifiableList(Lists.newArrayList(eqFieldIds)); Review Comment: nit: `ImmutableList.copyOf(eqFieldIds)`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
