[Enhancement] Limit memory used by ParallelIterable in Iceberg

[zhaohehuhu]

Why I'm doing:

The ConcurrentLinkedQueue doesn't have a size limitation in Iceberg 1.6.0. When the iceberg table is large, the queue will also become really large, which will cause OOM in the FE.

As per the above screenshot, the ConcurrentLinkedQueue is consuming 91%+ of the heap memory and causing the Frontend (FE) to go down, this indicates a significant memory management issue due to unbounded growth of the queue.

This issue was resolved with Iceberg 1.6.1(#10691)

What I'm doing:

as title

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 3.4
  • 3.3
  • 3.2
  • 3.1
  • 3.0

[zhaohehuhu]

@stephen-shelby @Youngwb plz help review

[github-actions]

[FE Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[stephen-shelby]

how many manifest files and data files in your case? I think if you use version 3.3 or above, this case may go to distributed plan, not local plan.

[zhaohehuhu]

how many manifest files and data files in your case? I think if you use version 3.3 or above, this case may go to distributed plan, not local plan.
Total Size of metadata is 439MB, while data files occupies 2TB space. Iceberg already released 1.6.1 to fix this issue, so it would be nice for us to update the version of Iceberg.

[stephen-shelby]

how many manifest files and data files in your case? I think if you use version 3.3 or above, this case may go to distributed plan, not local plan.
Total Size of metadata is 439MB, while data files occupies 2TB space. Iceberg already released 1.6.1 to fix this issue, so it would be nice for us to update the version of Iceberg.

could you try to set plan_mode=distributed then retry this query? you can observe if memory still high

[gengjun-git]

@Mergifyio rebase

[mergify]

rebase

✅ Branch has been successfully rebased

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[zhaohehuhu]

how many manifest files and data files in your case? I think if you use version 3.3 or above, this case may go to distributed plan, not local plan.
Total Size of metadata is 439MB, while data files occupies 2TB space. Iceberg already released 1.6.1 to fix this issue, so it would be nice for us to update the version of Iceberg.

could you try to set plan_mode=distributed then retry this query? you can observe if memory still high

OK. I will take a try.

[github-actions]

[Java-Extensions Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[github-actions]

[BE Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[zhaohehuhu]

Can we merge this pr firstly ? Iceberg has fixed some internal issues, so it’s fine to update the Iceberg version, just like Trino did.@stephen-shelby @gengjun-git

[zhaohehuhu]

set plan_mode=distributed

I failed to set plan_mode to distributed due to the issue(Failed to open the off-heap table scanner.)

[stephen-shelby]

set plan_mode=distributed

I failed to set plan_mode to distributed due to the issue(Failed to open the off-heap table scanner.)

you could check more detail msg in the fe.log.

[zhaohehuhu]

set plan_mode=distributed

I failed to set plan_mode to distributed due to the issue(Failed to open the off-heap table scanner.)

you could check more detail msg in the fe.log.

SQL Error [1064] [42000]: Failed to execute metadata collection job. Failed to open the off-heap table scanner. java exception details: java.lang.NoClassDefFoundError: Could not initialize class de.javakaffee.kryoserializers.UnmodifiableCollectionsSerializer
at com.starrocks.connector.iceberg.IcebergMetadataScanner.initSerializer(IcebergMetadataScanner.java:207)
at com.starrocks.connector.iceberg.IcebergMetadataScanner.open(IcebergMetadataScanner.java:132)

When the execution plan mode is switched to distributed, the issue occurs like above. This is a compatibility issue between JDK 17 and Kryo(may be fixed by #55016)

[zhaohehuhu]

closed. someone did it.