[Bug-4099][admin]bug fix 4 task with variable lineage

[MactavishCui]

Purpose of the pull request

#4099

Brief change log

Verify this pull request

This pull request is code cleanup without any test coverage.

(or)

This pull request is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

Before:

After:

[chenlb]

昨天也遇到了这个 bug，期待早点合并进来。

[aiwenmo]

Hi. Please refer to StudioServiceImpl.getLineage for the fix.
Before calculating the lineage, it includes five steps: databaseVariable, fragmentVariable, FlinkSQLEnv, RowPermission and UdfRefer.

[MactavishCui]

Hi. Please refer to StudioServiceImpl.getLineage for the fix. Before calculating the lineage, it includes five steps: databaseVariable, fragmentVariable, FlinkSQLEnv, RowPermission and UdfRefer.

Thx, I'll update this PR today.

[MactavishCui]

Hi. Please refer to StudioServiceImpl.getLineage for the fix. Before calculating the lineage, it includes five steps: databaseVariable, fragmentVariable, FlinkSQLEnv, RowPermission and UdfRefer.

Hi~ @aiwenmo I found that JobInstanceService's lineage build is based on execute History. Before a history being persisted to database, following methods are called:

create FlinkSqlTask -> FlinkSqlTask.getJobManager -> TaskService.buildJobSubmitConfig -> TaskService.buildEnvSql

which means databaseVariable, FlinkSQLEnv, RowPermission, FragmentVariable and UdfRefer are handled, then both statement and config field are updated. As a result we can reuse the config of History rather than create a new JobConfig or use default config like follows, I have also tested that this change can avoid the bug:

Before: return LineageBuilder.getColumnLineageByLogicalPlan(history.getStatement(), ExecutorConfig.DEFAULT);

After: return LineageBuilder.getColumnLineageByLogicalPlan(history.getStatement(), history.getConfigJson());

What is your opinion?
Eager to hear back.

[aiwenmo]

@MactavishCui I agree with you. Your discovery is wonderful.

[aiwenmo]

LGTM