Currently, in Flink I only support SupportsPrefixOperations IO. I believe we can add support for all IO types in a separate PR. If necessary, we can address it in this PR as well, but that would make things more complex.

This could be problematic if we have really big tables. In this case all of the files are read into a big List, which could cause OOM.

I don't really have a nice solution for this, but I think it is important to check if we have a better solution for this.

Maybe extract this from the operator, and provide this in the constructor.
It would be nice if this planner could remain independent of which table we read, and the query/table could be provided as a parameter

Do we need these non-consumer based public APIs?

we usually avoid get, set. See: https://iceberg.apache.org/contribute/#method-naming

Hmm... so we use Flink Java serialization?
Any better suggestion @mxm?

Or should we implement our own FileURI encoder instead of using Encoders.bean?

That's an interesting question to decide on.
We have metadata location Strings, and file system location Strings emitted from the downstream (FSList, MetadataList) operators. We need to serialize and shuffle them, so they matched by the path component, but need the scheme and the authority for matching, and uriAsString for deleting.

It is enough to have the uriAsString to travel on the wire. For the key, we need to get the path both on the emitter side, and on the AntiJoin side. It could be a fun exercise to check which performs better. OTOH the maintenance tasks are not the performance sensitive, so we can wait with the optimization until it is needed.

I would definitely opt for a Flink serializer for the FileURI class, and we can change it when it is needed. Also using Java serialization is one of the worst solution, so I would try to avoid it whenever it is not strictly needed. Especially that in this case FileURI should not change or the state will be corrupted.

How's this using Java serialization when the class here is not Serializable? If nothing is specified, Flink will try to use its PoJo serializer, but there are certain requirements like a no-arg constructor (https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/types_serialization/#pojos). Otherwise, we will fall back to the Kryo serialization framework, which is not the worst but of course a dedicated serializer is always going to be better.

My bad.. so Kryo, or own serializer then

Using this will make it awkward for the users, as they always have to use fully qualified name, because of the name conflicts. No better idea yet, so just writing down for now.

Why do we need to use fully qualified name here? Which other class clashes with PrefixMismatchMode?

If the restriction is removed, it will be DeleteOrphanFiles.PrefixMismatchMode. The name DeleteOrphanFiles conflicts with a class in Flink.

This compiles for me:

import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
...
    private PrefixMismatchMode prefixMismatchMode = PrefixMismatchMode.ERROR;
...
    public Builder prefixMismatchMode(PrefixMismatchMode newPrefixMismatchMode) {
      this.prefixMismatchMode = newPrefixMismatchMode;
      return this;
    }

Then, from another class, this works as well:

import org.apache.iceberg.actions.DeleteOrphanFiles.PrefixMismatchMode;
...
      DeleteOrphanFiles.builder().prefixMismatchMode(PrefixMismatchMode.IGNORE);

Thank you very much, this is indeed feasible.

Could you please check when would we need this? Seems strange, and I don't understand why would the user need to set this.

To use IcebergSourceSplitSerializer, the constructor requires.So I have to exposed to user.

Could you please dig a bit deeper so we understand why the IcebergSourceSplitSerializer needs the caseSensitive flat, and how it is used?

In IcebergSourceSplit deserialization, caseSensitive needs to be passed down to the underlying layer. I’m not sure if this is to maintain compatibility with different source JSON systems or to support historical data formats. In toJson, caseSensitive is not required, but in fromJson, it needs to be passed.

IcebergSourceSplit.deserializeV3(serialized, caseSensitive);
FileScanTask task = ScanTaskParser.fromJson(taskJson, caseSensitive);

You can follow the usage of the caseSensitive in the fromJson method, and you can find, that it is used when deserializing the FileScanTask. Specifically when creating the filter (ResidualEvaluator residualEvaluator = ResidualEvaluator.of(spec, filter, caseSensitive);)

ScanContext already contains caseSensitive. We can reuse it instead of requesting it again.

You are right, I will do it !

How do we get here if the 2 authorities are not exactly the same?

Ok.. so the keyBy is done on the getPath. What happens if we have multiple versions in the metadata for the given path? Shall we check that too?

I understand that the scenario where multiple versions exist for a given path should only occur when the authority changes.

In HDFS, authority changes would happen when foundInFileSystem detects a change (like a namenode-host change), and in that case, those files should not be deleted.
In S3, if the authority changes, I understand it happens when switching between different S3 endpoints or regions, and those files should also not be deleted.
I am mainly following Spark’s logic here:

Please let me know if there are any gaps or misunderstandings in my reasoning. Thank you!

Very edge case, but let's say that the authority changes, so we have multiple versions of the same path in the table metadata. Also let's say that the equalAuthorities is configured incorrectly. In this case the result will be different based on the order the files are received by the AntiJoin.

Maybe when we receive the 2nd element from the metadata, we could immediately check if the 2 values are matching or, not.

Would it worth the complexity to start dropping filesToDelete immediately after we get an exception?

The two streams are unordered. If an exception occurs in processElement2 and filesToDelete is cleared immediately, there is no guarantee that filesToDelete will be empty when processWatermark is called.

What I meant:

• If there is an error, then filesToDelete.empty(), and set a boolean flag, that new values are also dropped.
• We still keep the exception in state, and handling the watermark is the same, or just emit all filesToDelete, since they are empty anyway when there is an error

This is basically removes the generality of the TableReader. We extract the value of the first column of the row, and emit it as a string.
Do we want to create a parent class as TableReader<R> which gets a parameter (rowData, Collector<R>)->{} method to extract the values, or call this class differently, like FileNameReader for now, and do this if we reuse it in the ManifestRewrite flow?

I think we can first implement an initial version according to Plan One, and then make adjustments based on the specific situation when rewriting the manifest.

What is Plan One in this case? 😄

Create a parent class as TableReader which gets a parameter (rowData, Collector)->{} method to extract the values

Ok. Sounds good. Go ahead and do this.

This is kind of arbitrary. Should we bump this to a week at least?

The 3-day configuration is aligned with the Spark part. https://iceberg.incubator.apache.org/docs/nightly/spark-procedures/#remove_orphan_files. So I keep the same value.

Can we pull the default from an existing value for the default worker pool size?

We can use ThreadPools.WORKER_THREAD_POOL_SIZE to install of use a number . Change it .

Why is this set so low by default?

This configuration is aligned with the Spark part. So I keep the same value here.

Would it make sense to limit the total number of files instead of the subdirectories?

1. Limiting ListingDirectSubDirs and ListingDepth is provided for distributed tasks. However, if we use MaxFilesCount, in my view, we can't further refine the tasks.
2. On the Spark side, the above approach is used to distribute the folders that cannot be processed from the driver to the workers for distributed searching. But currently, the Flink side implementation is only within a single operator, so I don't think there is much impact here for now.
   WDYT？

I don't understand (1). Why does a max file count limit the refinement? Doesn't the same apply to filtering by directory depth and number of subdirectories?

Concerning (2), wouldn't we benefit in Flink from processing a limited number of files per table maintenance? Perhaps, this should be a global option, rather than a DeleteOrphan files option.

I understand what you mean. If we add a maximum file limit here, it indeed can shorten the execution time of a task, which is equivalent to checking whether some system files are orphan files. After thinking it through, I believe it is necessary. If we are to add this configuration item, I think it should be added at the core level.

Can we not support this feature in this PR, but instead open another PR to design and implement it?

I would be a bit hesitant on limiting the number of files listed by the DeleteOrphanFiles.
The listing needs to list all of the files in the table directory. If we limit the number of files returned, then it would cause erratic behavior - the job runs and succeeds, but the orphan files are not removed. But some cases they might be removed depending on the listing.

If we fear that the listing is too big for a single task, we could use the Spark approach, and distribute the remaining dirs to a downstream operator. I believe that this is a worthwhile improvement, but probably worth to postpone to the next PR

@pvary @mxm I think it's feasible to distribute the remaining directories to a downstream operator, and we can do that in the next PR.
Are there any other issues here that we need to address? Can we move forward with this pr?
Thanks!

Sounds good @Guosmilesmile, but could you explain why we need the limits for listing depth and direct subdirectories? I understand that we want to have some safeguards, but why are these limits so low?

@mxm Because these two configurations can be set by users, I aligned the default values with the Spark part when setting them.

Can we add a comment to highlight where those defaults are coming from?

Ok, I will do it now !

nit: Is this originalTable?
It is usually better to not use abbreviations

Fix it .